hello-agents/Co-creation-projects/lcyting-StockSage-agent/skills/金融数据/mx-data/mx_data.py

#!/usr/bin/env python3
# mx_data - 妙想金融数据查询 skill
# 基于东方财富妙想API提供金融数据查询能力
# 输出 Excel（多sheet）+ 描述 txt

import os
import sys
import json
import re
import pandas as pd
import requests
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple

def safe_filename(s: str, max_len: int = 80) -> str:
    """Convert query string to safe filename """
    s = re.sub(r'[<>:"/\\|?*\[\]]', "_", s)
    s = s.strip().replace(" ", "_")[:max_len]
    return s or "query"

def flatten_value(v: Any) -> str:
    """Flatten any value to string """
    if v is None:
        return ""
    if isinstance(v, (dict, list)):
        return json.dumps(v, ensure_ascii=False)
    return str(v)

def ordered_keys(table: Dict[str, Any], indicator_order: List[Any]) -> List[Any]:
    """Order keys """
    data_keys = [k for k in table.keys() if k != "headName"]
    key_map = {str(k): k for k in data_keys}
    preferred: List[Any] = []
    seen: set[str] = set()
    for key in indicator_order:
        key_str = str(key)
        if key_str in key_map and key_str not in seen:
            preferred.append(key_map[key_str])
            seen.add(key_str)
    for key in data_keys:
        key_str = str(key)
        if key_str not in seen:
            preferred.append(key)
            seen.add(key_str)
    return preferred

def normalize_values(raw_values: List[Any], expected_len: int) -> List[str]:
    """Normalize values length """
    values = [flatten_value(v) for v in raw_values]
    if len(values) < expected_len:
        values.extend([""] * (expected_len - len(values)))
    return values[:expected_len]

def return_code_map(block: Dict[str, Any]) -> Dict[str, str]:
    """Get return code map"""
    for key in ("returnCodeMap", "returnCodeNameMap", "codeMap"):
        data = block.get(key)
        if isinstance(data, dict):
            return {str(k): flatten_value(v) for k, v in data.items()}
    return {}

def format_indicator_label(key: str, name_map: Dict[str, Any], code_map: Dict[str, str]) -> str:
    """Format indicator label"""
    mapped = name_map.get(key)
    if mapped is None and key.isdigit():
        mapped = name_map.get(int(key))
    if mapped not in (None, ""):
        return flatten_value(mapped)
    mapped_code = code_map.get(key)
    if mapped_code not in (None, ""):
        return flatten_value(mapped_code)
    if key.isdigit():
        return ""
    return key

def table_to_rows(block: Dict[str, Any]) -> Tuple[List[Dict[str, str]], List[str]]:
    """
    Convert table to rows - adapted for MX API format:
    MX API format: headName is date array, each indicator is an array matching headName length
    This is DATE-ON-ROW format, each row is a date
    """
    table = block.get("table") or {}
    name_map = block.get("nameMap") or {}
    if isinstance(name_map, list):
        name_map = {str(i): v for i, v in enumerate(name_map)}
    elif not isinstance(name_map, dict):
        name_map = {}

    if not isinstance(table, dict):
        # Fallback for generic tables
        if isinstance(table, list):
            if not table:
                return [], []
            if isinstance(table[0], dict):
                rows = table
            else:
                rows = [
                    dict(zip([f"column_{i}" for i in range(len(table[0]))], row))
                    for row in table
                ]
        else:
            return [], []
        return [{name_map.get(k, k): flatten_value(v) for k, v in row.items()} for row in rows], list(rows[0].keys())

    headers = table.get("headName") or []
    if not isinstance(headers, list):
        headers = []
    order = ordered_keys(table, block.get("indicatorOrder") or [])
    entity_name = flatten_value(block.get("entityName") or "") or "指标"
    code_map = return_code_map(block)

    rows: List[Dict[str, str]] = []
    data_key_count = len([key for key in table.keys() if key != "headName"])

    # MX API format: headName is the date column (one row = one date)
    # headName = [date1, date2, date3, ...]
    # indicator = [val1, val2, val3, ...]
    # So each row is a date with all indicator values
    if len(headers) > 0:  # has dates (one date per row)
        fieldnames = ["date"]  # first column is date from headName
        for key in order:
            if key != "headName":
                label = format_indicator_label(str(key), name_map, code_map)
                if label:  # only add non-empty labels
                    fieldnames.append(label)
        
        # Build one row per date
        for row_idx, date in enumerate(headers):
            row = {"date": flatten_value(date)}
            for key in order:
                if key == "headName":
                    continue
                label = format_indicator_label(str(key), name_map, code_map)
                if not label:
                    continue
                raw_values = table.get(key, [])
                value = raw_values[row_idx] if row_idx < len(raw_values) else ""
                row[label] = flatten_value(value)
            rows.append(row)
        
        return rows, fieldnames

    if len(headers) == 1 and data_key_count >= 1:
        # Single date (current quote)
        fieldnames = [entity_name, flatten_value(headers[0])]
        for key in order:
            raw_values = table.get(key, [])
            value = raw_values[0] if isinstance(raw_values, list) and raw_values else raw_values
            label = format_indicator_label(str(key), name_map, code_map)
            rows.append({fieldnames[0]: label, fieldnames[1]: flatten_value(value)})
        return rows, fieldnames

    # headName 为空但各指标为等长数组：按索引展开成行（十大股东等无日期维度场景）
    if len(headers) == 0 and data_key_count >= 1:
        col_labels: List[str] = []
        col_arrays: List[List[Any]] = []
        for key in order:
            if key == "headName":
                continue
            raw_values = table.get(key, [])
            if not isinstance(raw_values, list):
                raw_values = [raw_values] if raw_values not in (None, "") else []
            lab = format_indicator_label(str(key), name_map, code_map)
            if not lab:
                lab = str(key)
            col_labels.append(lab)
            col_arrays.append(raw_values)
        if col_labels:
            n_rows = max((len(a) for a in col_arrays), default=0)
            if n_rows > 0:
                wide_rows: List[Dict[str, str]] = []
                for i in range(n_rows):
                    row_d: Dict[str, str] = {}
                    for lab, arr in zip(col_labels, col_arrays):
                        row_d[lab] = flatten_value(arr[i]) if i < len(arr) else ""
                    wide_rows.append(row_d)
                return wide_rows, col_labels

    # Fallback
    return [], []

class MXData:
    """妙想金融数据查询客户端"""
    
    BASE_URL = "https://mkapi2.dfcfs.com/finskillshub/api/claw/query"
    
    def __init__(self, api_key: Optional[str] = None):
        """
        初始化客户端
        :param api_key: MX API Key，如果不提供则从环境变量 MX_APIKEY 读取
        """
        self.api_key = api_key or os.getenv("MX_APIKEY")
        if not self.api_key:
            raise ValueError(
                "MX_APIKEY 环境变量未设置，请先设置环境变量：\n"
                "export MX_APIKEY=your_api_key_here\n"
                "或者在初始化时传入 api_key 参数"
            )
    
    def query(self, tool_query: str) -> Dict[str, Any]:
        """
        查询金融数据
        :param tool_query: 自然语言查询问句，如 "东方财富最新价"
        :return: API 响应结果
        """
        headers = {
            "Content-Type": "application/json",
            "apikey": self.api_key
        }
        data = {
            "toolQuery": tool_query
        }
        
        response = requests.post(self.BASE_URL, headers=headers, json=data, timeout=30)
        response.raise_for_status()
        return response.json()
    
    @staticmethod
    def parse_result(result: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], List[str], int, Optional[str]]:
        """
        Parse API result into tables for Excel output
        :return: (tables, condition_parts, total_rows, error)
            tables: [{"sheet_name": str, "rows": list[dict], "fieldnames": list[str]}]
        """
        status = result.get("status")
        message = result.get("message", "")
        if status != 0:
            return [], [], 0, f"顶层错误: 状态码 {status} - {message}"
        
        data = result.get("data", {})
        inner_data = data.get("data", {})
        search_result = inner_data.get("searchDataResultDTO", {})
        dto_list = search_result.get("dataTableDTOList", [])
        
        if not dto_list:
            return [], [], 0, "接口返回中无 dataTableDTOList"
        
        condition_parts: List[str] = []
        tables: List[Dict[str, Any]] = []
        total_rows = 0
        
        for i, dto in enumerate(dto_list):
            if not isinstance(dto, dict):
                continue
            
            sheet_name = safe_filename(
                dto.get("title") or dto.get("inputTitle") or dto.get("entityName") or f"表{i + 1}"
            )
            condition = dto.get("condition")
            if condition is not None and condition != "":
                entity = dto.get("entityName") or sheet_name
                condition_parts.append(f"[{entity}]\n{condition}")
            
            rows, fieldnames = table_to_rows(dto)
            if not rows:
                raw_tbl = dto.get("rawTable")
                if isinstance(raw_tbl, dict) and raw_tbl:
                    alt = dict(dto)
                    alt["table"] = raw_tbl
                    rows, fieldnames = table_to_rows(alt)
            if not rows:
                continue
            tables.append({"sheet_name": sheet_name, "rows": rows, "fieldnames": fieldnames})
            total_rows += len(rows)
        
        if not tables:
            return [], condition_parts, 0, "dataTableDTOList 中无有效 table 数据"
        return tables, condition_parts, total_rows, None
    
    @staticmethod
    def format_terminal(result: Dict[str, Any], tables: List[Dict[str, Any]], total_rows: int) -> str:
        """Format result for terminal display"""
        output = []
        status = result.get("status")
        message = result.get("message", "")
        if status != 0:
            output.append(f"**错误**: 状态码 {status} - {message}")
            return "\n".join(output)
        
        data = result.get("data", {})
        inner_data = data.get("data", {})
        search_result = inner_data.get("searchDataResultDTO", {})
        entity_tags = search_result.get("entityTagDTOList", [])
        
        if entity_tags:
            output.append("**查询证券**:")
            entities = []
            for tag in entity_tags:
                name = tag.get("fullName", "")
                code = tag.get("secuCode", "")
                type_name = tag.get("entityTypeName", "")
                entities.append(f"- {name} ({code}) - {type_name}")
            output.append("\n".join(entities))
            output.append("")
        
        output.append(f"**查询结果**: {len(tables)} 个表，共 {total_rows} 行数据\n")
        
        # Only preview first 20 rows of first table on terminal
        if tables:
            first = tables[0]
            output.append(f"**{first['sheet_name']}** (前20行预览):\n")
            rows = first["rows"][:20]
            if rows:
                fieldnames = first["fieldnames"]
                output.append("| " + " | ".join(fieldnames) + " |")
                output.append("| " + " | ".join(["---"] * len(fieldnames)) + " |")
                for row in rows:
                    cells = [str(row.get(f, "")) for f in fieldnames]
                    output.append("| " + " | ".join(cells) + " |")
                if len(first["rows"]) > 20:
                    output.append(f"| ... | " * (len(fieldnames) - 1) + "... |")
                output.append("")
        
        question_id = search_result.get("questionId", "")
        if question_id:
            output.append(f"*查询ID: {question_id}*")
        
        return "\n".join(output)
    
    @staticmethod
    def write_output_files(
        query_text: str,
        output_dir: Path,
        tables: List[Dict[str, Any]],
        total_rows: int,
        condition_parts: List[str],
    ) -> Tuple[Path, Path]:
        """Write output Excel (multiple sheets) and description"""
        safe_name = safe_filename(query_text)
        file_path = output_dir / f"mx_data_{safe_name}.xlsx"
        desc_path = output_dir / f"mx_data_{safe_name}_description.txt"
        
        # Write all tables to a single Excel file with multiple sheets
        with pd.ExcelWriter(file_path, engine="openpyxl") as writer:
            for table in tables:
                df = pd.DataFrame(table["rows"], columns=table["fieldnames"])
                df.to_excel(writer, sheet_name=table["sheet_name"], index=False)
        
        # Write description file
        description_lines = [
            "金融数据查询结果说明",
            "=" * 40,
            f"查询内容: {query_text}",
            f"数据文件路径: {file_path}",
            f"描述文件路径: {desc_path}",
            f"数据行数: {total_rows}",
            f"表数量: {len(tables)}",
            f"Sheet 列表: {', '.join([t['sheet_name'] for t in tables])}",
        ]
        if condition_parts:
            description_lines.append("")
            description_lines.append("筛选条件:")
            description_lines.extend(condition_parts)
        
        desc_path.write_text("\n".join(description_lines), encoding="utf-8")
        return file_path, desc_path

def main():
    """命令行入口 - 保持与 MX_FinData 一致的使用方式，输出 Excel 多 sheet"""
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} \"查询问句\" [输出目录]")
        print(f"默认输出目录: /root/.openclaw/workspace/mx_data/output/")
        print("示例: python mx_data.py \"同花顺最近3年每天的最新价\"")
        sys.exit(1)
    
    # Build query
    if len(sys.argv) >= 3:
        query = " ".join(sys.argv[1:-1])
        output_dir = Path(sys.argv[-1])
    else:
        query = " ".join(sys.argv[1:])
        # Default output to fixed directory
        output_dir = Path("/root/.openclaw/workspace/mx_data/output")
    
    # Ensure output directory exists
    output_dir.mkdir(parents=True, exist_ok=True)
    
    try:
        mx = MXData()
        result = mx.query(query)
        tables, condition_parts, total_rows, err = mx.parse_result(result)
        
        if err:
            print(f"错误: {err}")
            sys.exit(1)
        
        # Terminal preview
        print(mx.format_terminal(result, tables, total_rows))
        
        # Write Excel (multiple sheets) + description txt
        file_path, desc_path = mx.write_output_files(query, output_dir, tables, total_rows, condition_parts)
        print(f"\n✅ Excel 文件: {file_path}")
        print(f"📄 描述文件: {desc_path}")
        print(f"📊 总行数: {total_rows}, 表数: {len(tables)}")
        
        # Save original JSON
        json_filename = output_dir / f"mx_data_{safe_filename(query)}_raw.json"
        with open(json_filename, "w", encoding="utf-8") as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
        print(f"📄 原始JSON: {json_filename}")
            
    except Exception as e:
        print(f"错误: {str(e)}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()