hello-agents/Co-creation-projects/lcyting-StockSage-agent/backend/app/services/market_service.py

"""
智能股票分析助手 — 行情数据服务层

封装金融数据查询、解析和格式化逻辑，供API路由层调用。
含妙想 mx_data 计时缓存与额度用尽时的缓存降级。
"""

from __future__ import annotations

import copy
import math
import re
import sys
from pathlib import Path
from typing import Any, Optional

# 确保skills路径可导入
_PROJECT_ROOT = Path(__file__).parent.parent.parent.parent  # backend/app/services -> project root
_AGENTS_DIR = _PROJECT_ROOT / "agents"
_SKILLS_DATA = _PROJECT_ROOT / "skills" / "金融数据" / "mx-data"

for p in [_AGENTS_DIR, _SKILLS_DATA, str(_PROJECT_ROOT)]:
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))

from app.config import settings
from app.services.mx_timed_cache import get_mx_timed_cache, mx_cache_ttl_seconds
from app.utils.mx_fixture import try_load_raw_fixture
from app.utils.mx_quota import MX_QUOTA_HINT, is_mx_quota_exhausted, quota_exhausted_no_cache_message

# 仪表盘指数卡片：从 mx tables 中解析列名（妙想返回的表头差异较大）
_PRICE_HDR = re.compile(
    r"点位|最新点|收盘点|指数点位|收盘价|最新价|现价|收盘|价格|数值|行情|昨收|今开|当前价|最新报价|报价",
    re.I,
)
_CHANGE_HDR = re.compile(
    r"涨跌幅|涨跌幅度|当日涨幅|日涨跌幅|涨跌|涨幅|变动率",
    re.I,
)
_DATE_HDR = re.compile(r"日期|时间|^date$", re.I)
_LONG_PRICE_LABEL = re.compile(
    r"点位|最新点|收盘点|指数点位|收盘价|最新价|现价|收盘|价格|最新|上证|深证|成指|沪深300|创业板指",
)
_LONG_CHANGE_LABEL = re.compile(r"涨跌幅|涨跌幅度|当日涨幅|涨跌|涨幅|变动率")


def _parse_pct_cell(val: Any) -> Optional[float]:
    """解析涨跌幅单元格为浮点数（百分比数值，不带 % 也可）"""
    if val is None:
        return None
    if isinstance(val, (int, float)):
        x = float(val)
        return x if math.isfinite(x) else None
    s = str(val).strip()
    if not s or s in ("--", "—", "-", "暂无"):
        return None
    s = re.sub(r"[%％,，]", "", s)
    s = s.replace("＋", "+").replace("−", "-").replace("－", "-")
    m = re.search(r"[-+]?\d+(?:\.\d+)?(?:e[-+]?\d+)?", s, re.I)
    if not m:
        return None
    try:
        x = float(m.group(0))
    except ValueError:
        return None
    return x if math.isfinite(x) else None


def _cell_at(names: list, row: dict, idx: int) -> Any:
    if idx < 0 or not isinstance(row, dict):
        return None
    if idx >= len(names):
        return None
    return row.get(names[idx])


def _heuristic_index_from_row(row: dict, names: list[str]) -> tuple[Optional[str], Optional[float]]:
    """列名无法识别时，按数值量级与 % 符号兜底"""
    change: Optional[float] = None
    price_disp: Optional[str] = None
    best: float = -1.0
    keys = [n for n in names if n in row] if names else list(row.keys())
    for k in keys:
        if _DATE_HDR.search(str(k)):
            continue
        raw = row[k]
        if raw is None:
            continue
        sv = str(raw).strip()
        if not sv:
            continue
        if "%" in sv or "％" in sv:
            p = _parse_pct_cell(sv)
            if p is not None:
                change = p
            continue
        p = _parse_pct_cell(sv)
        if p is None:
            continue
        # A 股主要指数点位多在数百～数万之间
        if 500 <= abs(p) <= 50000:
            if abs(p) > best:
                best = abs(p)
                price_disp = sv
    return price_disp, change


def _extract_index_card_one_table(t0: dict) -> tuple[Optional[str], Optional[float]]:
    """从 mx_data 单个 sheet 解析点位与涨跌幅（字段名与行结构随品种变化较大）"""
    names: list = list(t0.get("fieldnames") or t0.get("fieldNames") or [])
    rows = t0.get("rows") or []
    if not rows:
        return None, None

    # 长表：恰两列，每行一个指标
    if len(names) == 2:
        lk, vk = names[0], names[1]
        price_s: Optional[str] = None
        change_v: Optional[float] = None
        for r in rows:
            if not isinstance(r, dict):
                continue
            lab = str(r.get(lk, "")).strip()
            raw = r.get(vk)
            if _LONG_CHANGE_LABEL.search(lab):
                c = _parse_pct_cell(raw)
                if c is not None:
                    change_v = c
            elif _LONG_PRICE_LABEL.search(lab):
                if price_s is None and raw is not None and str(raw).strip():
                    price_s = str(raw).strip()
        if price_s or change_v is not None:
            return price_s, change_v

    date_idx = next((i for i, n in enumerate(names) if _DATE_HDR.search(str(n))), -1)
    data_row = rows[-1] if date_idx >= 0 and len(rows) > 1 else rows[0]

    if isinstance(data_row, list):
        pi = next((i for i, n in enumerate(names) if _PRICE_HDR.search(str(n))), -1)
        ci = next((i for i, n in enumerate(names) if _CHANGE_HDR.search(str(n))), -1)
        raw_p = data_row[pi] if 0 <= pi < len(data_row) else None
        raw_c = data_row[ci] if 0 <= ci < len(data_row) else None
        ps = str(raw_p).strip() if raw_p is not None and str(raw_p).strip() else None
        cv = _parse_pct_cell(raw_c)
        if ps or cv is not None:
            return ps, cv
        return None, None

    if isinstance(data_row, dict):
        pi = next((i for i, n in enumerate(names) if _PRICE_HDR.search(str(n))), -1)
        ci = next((i for i, n in enumerate(names) if _CHANGE_HDR.search(str(n))), -1)
        raw_p = _cell_at(names, data_row, pi)
        raw_c = _cell_at(names, data_row, ci)
        ps = str(raw_p).strip() if raw_p is not None and str(raw_p).strip() else None
        cv = _parse_pct_cell(raw_c)
        if ps or cv is not None:
            return ps, cv
        return _heuristic_index_from_row(data_row, names)

    return None, None


def _extract_index_card_from_tables(tables: list) -> tuple[Optional[str], Optional[float]]:
    """从 mx_data 全部 sheet 解析仪表盘展示字段（妙想常返回多表，首表可能是说明性空表）"""
    if not tables:
        return None, None
    merged_p: Optional[str] = None
    merged_c: Optional[float] = None
    for t in tables:
        if not isinstance(t, dict):
            continue
        p, c = _extract_index_card_one_table(t)
        if p is not None and c is not None:
            return p, c
        if merged_p is None and p is not None:
            merged_p = p
        if merged_c is None and c is not None:
            merged_c = c
        if merged_p is not None and merged_c is not None:
            return merged_p, merged_c
    return merged_p, merged_c


def _enrich_index_quote_result(result: dict) -> dict:
    """成功返回时附加 display_*，供前端直接使用；deepcopy 避免污染进程内缓存对象"""
    if not result.get("success"):
        return result
    out = copy.deepcopy(result)
    price, chg = _extract_index_card_from_tables(out.get("tables") or [])
    if price is not None:
        out["display_price"] = price
    if chg is not None:
        out["display_change_pct"] = chg
    return out


def _mx_result_meta(*, from_cache: bool, quota_exhausted: bool, channel: str) -> dict:
    ttl = int(mx_cache_ttl_seconds())
    m = {
        "from_cache": from_cache,
        "quota_exhausted": quota_exhausted,
        "cache_ttl_seconds": ttl,
        "channel": channel,
    }
    if quota_exhausted:
        m["hint"] = MX_QUOTA_HINT
    return m


def _attach_meta(payload: dict, meta: dict) -> dict:
    out = copy.deepcopy(payload)
    out["_mx_meta"] = meta
    return out


def _fetch_mx_data_live(query: str) -> dict:
    """直连妙想 mx_data；MX_REPLAY_FIXTURES 时优先读本地原始 JSON（不修额度）"""
    import mx_data as _mx

    result = {
        "success": False,
        "query": query,
        "tables": [],
        "condition_parts": [],
        "total_rows": 0,
        "error": None,
    }

    raw_fixture = try_load_raw_fixture("mx_data", query)
    if raw_fixture is not None:
        try:
            tables, condition_parts, total_rows, error = _mx.MXData.parse_result(raw_fixture)
            if error:
                result["error"] = f"[fixture] {error}"
                return result
            result["success"] = True
            result["tables"] = tables
            result["condition_parts"] = condition_parts
            result["total_rows"] = total_rows
            return result
        except Exception as e:
            result["error"] = f"[fixture] 解析失败: {e}"
            return result

    key_ok = bool(settings.MX_APIKEY and settings.MX_APIKEY != "your-mx-apikey-here")
    if not key_ok:
        result["error"] = "MX_APIKEY 未配置，且无匹配的本地 fixture（设置 MX_REPLAY_FIXTURES=1 并放置 JSON）"
        return result

    try:
        querier = _mx.MXData(api_key=settings.MX_APIKEY)
        raw_result = querier.query(query)
        tables, condition_parts, total_rows, error = _mx.MXData.parse_result(raw_result)

        if error:
            result["error"] = error
            return result

        result["success"] = True
        result["tables"] = tables
        result["condition_parts"] = condition_parts
        result["total_rows"] = total_rows
        return result

    except Exception as e:
        result["error"] = str(e)
        return result


def query_financial_data(query: str) -> dict:
    """执行金融数据查询并返回结构化结果（带计时缓存与额度降级）

    缓存键为规范化后的自然语言 query：
    - 相同查询串在 TTL 内不会重复请求妙想；
    - 股票代码 / 财务指标不同会得到不同 query，从而自动区分。
    """
    result = {
        "success": False,
        "query": query,
        "tables": [],
        "condition_parts": [],
        "total_rows": 0,
        "error": None,
    }

    key_missing = not settings.MX_APIKEY or settings.MX_APIKEY == "your-mx-apikey-here"
    if key_missing and not settings.MX_REPLAY_FIXTURES:
        result["error"] = "MX_APIKEY 未配置"
        return result

    cache = get_mx_timed_cache()
    ttl = mx_cache_ttl_seconds()
    key = cache.make_key("mx_data", query)

    fresh = cache.get_fresh(key, ttl)
    if fresh is not None:
        return _attach_meta(
            fresh,
            _mx_result_meta(from_cache=True, quota_exhausted=False, channel="mx_data"),
        )

    live = _fetch_mx_data_live(query)

    if live["success"]:
        cache.set(key, live)
        return _attach_meta(
            live,
            _mx_result_meta(from_cache=False, quota_exhausted=False, channel="mx_data"),
        )

    err = live.get("error") or ""
    if is_mx_quota_exhausted(err):
        stale = cache.get_stale(key)
        if stale:
            merged = copy.deepcopy(stale)
            merged["success"] = True
            merged["query"] = query
            return _attach_meta(
                merged,
                _mx_result_meta(from_cache=True, quota_exhausted=True, channel="mx_data"),
            )
        live["error"] = quota_exhausted_no_cache_message(err)
        return live

    return live


def get_stock_quote(code: str) -> dict:
    """查询个股实时行情

    一并请求 OHLC/昨收，供前端「当日价位快照」图使用；仅查最新价时五价合一易变成一条平线。
    优先从文件缓存读取，未命中或过期才调用接口。
    """
    from app.services.stock_file_cache import get_stock_file_cache
    fc = get_stock_file_cache()

    cached = fc.get(code, "quote")
    if cached and cached.get("data"):
        cached_data = cached["data"]
        if cached_data.get("success"):
            return cached_data

    extra = "今开 开盘 最高 最低 昨收 昨收盘价"
    if code.startswith(("6", "5", "9")):
        query = f"{code} 最新价 涨跌幅 涨跌额 {extra} 成交量 成交额 换手率"
    else:
        query = f"{code} 最新价 涨跌幅 涨跌额 {extra} 成交量 成交额 换手率"

    result = query_financial_data(query)
    if result.get("success"):
        fc.set(code, "quote", result)
    return result


def get_stock_financial(code: str, indicators: str = "净利润 营业收入 净资产收益率 每股收益") -> dict:
    """查询个股财务指标（文件缓存优先）"""
    from app.services.stock_file_cache import get_stock_file_cache
    fc = get_stock_file_cache()

    cached = fc.get(code, "financial")
    if cached and cached.get("data") and cached["data"].get("success"):
        return cached["data"]

    query = f"{code} {indicators}"
    result = query_financial_data(query)
    if result.get("success"):
        fc.set(code, "financial", result)
    return result


def get_stock_profile(code: str) -> dict:
    """查询公司概况（文件缓存优先）"""
    from app.services.stock_file_cache import get_stock_file_cache
    fc = get_stock_file_cache()

    cached = fc.get(code, "profile")
    if cached and cached.get("data") and cached["data"].get("success"):
        return cached["data"]

    query = f"{code} 公司简介 主营业务 成立时间 董事长 总股本"
    result = query_financial_data(query)
    if result.get("success"):
        fc.set(code, "profile", result)
    return result


def get_stock_holders(code: str) -> dict:
    """查询十大股东（文件缓存优先）"""
    from app.services.stock_file_cache import get_stock_file_cache
    fc = get_stock_file_cache()

    cached = fc.get(code, "holders")
    if cached and cached.get("data") and cached["data"].get("success"):
        return cached["data"]

    query = f"{code} 十大股东"
    result = query_financial_data(query)
    if result.get("success"):
        fc.set(code, "holders", result)
    return result


def get_index_quote(index_name: str = "沪深300") -> dict:
    """查询指数行情（附带 display_price / display_change_pct 供仪表盘稳定展示）"""
    # 避免「上证指数指数」重复；自然语言尽量简短明确
    query = f"{index_name} 最新点位 涨跌幅"
    base = query_financial_data(query)
    return _enrich_index_quote_result(base)


def get_sector_quote(sector_name: str) -> dict:
    """查询板块行情"""
    query = f"{sector_name}板块最新行情"
    return query_financial_data(query)