akg-factor-bridge/sources.py

"""候选卡的取数层：把各条证据线从三处库读成"按前缀码索引的字典"（全部只读）。

## 为什么要有它

候选卡的规则在 card.py，是纯函数；证据从哪来、怎么对齐代码格式、缺了怎么办，
全部收在这里，plan.py 只做装配。三处来源：

  基座 PG   v_factor_transmission_moved   已启动成员及其所在环节的传导证据（只认 Segment 目标）
            v_factor_stock_daily          数据日涨幅、主力净额异常值、热度变化
  153 代理  strategy_daily_results        决策系统昨夜结论：信号、支撑压力位、吸筹块
            （吸筹的评分、状态、评分日三项一次读齐；基座落库的吸筹版本没有评分日，所以不从基座取）
  平台 MySQL gp_day_data                  交易日历（算评分日龄用；取一只长期存在的票的日期序列）
  基座 PG   v_factor_logic                数据基座抽取的因果论断（方向、机制、时效、出处），每票最多几条，
                                          只展示不作门槛（2026-09-03 方案第 3.3 节）
  平台 MySQL zs_day_data / eastmoney_rzrq_data / fear_greed_index
                                          计划环境段的市场四项：两市成交额、融资余额、恐贪指数；
                                          市场广度从基座 v_factor_stock_daily 当日行自算（同一节）

代码格式：基座是点后缀式 600000.SH，决策系统与桥是前缀式 SH600000，进出都过 common.to_prefix。
读失败的语义：每一路读不到都返回空字典并打印一行原因，候选卡按"缺失"处理（进关注或仅展示），
不让计划断产——与 pool.py 的安全边界一致。

离线单测：logic_claims 与 market_context 都接受注入的读函数（read_pg / read_mysql），
test_market_context.py 用假数据函数替换真实连接，不连库；两者内部只对"记录列表"做计算，
读函数返回 DataFrame 或普通的字典列表都可以（见 _records）。
"""
from __future__ import annotations

import datetime as dt
import json
import statistics

import pandas as pd

import common
import config
import db

# 上证指数与深证成指在指数日线表 zs_day_data 里的代码，两市成交额取二者当日 amount 之和。
MARKET_INDEX_CODES = ("000001.SH", "399001.SZ")
# 市场广度里"涨停家数"的近似口径：涨幅达到 9.8%（不读涨跌停价表，与环节日行情视图注释一致）。
LIMIT_UP_PCT = 9.8
# 融资表与恐贪指数表的日期列名不在本仓库内核实过，按候选名逐个匹配（第一个命中的用）。
_DATE_COL_CANDIDATES = ("trade_date", "date", "stat_date", "data_date", "report_date",
                        "dt", "timestamp", "update_date", "record_date", "created_at")


def moved_members(ds: str) -> dict[str, dict]:
    """数据日 ds 被传导指向的环节里，已启动（不在未动名单）的成员。
    同一票挂在多个被指向环节上时，取源数最多、其次链符最高的那条作卡上的证据。"""
    try:
        df = db.read_pg(
            "SELECT ts_code, target, n_sources, chain_fit, members_total, moved, "
            "moved_ratio, mkt_trade_date FROM v_factor_transmission_moved "
            "WHERE scan_date = %s", (ds,))
    except Exception as e:  # noqa: BLE001
        print(f"  （已动成员视图读取失败，候选卡的传导门槛整体缺席: {e!r}）")
        return {}
    if df.empty:
        return {}
    df["k"] = df["ts_code"].map(lambda s: common.to_prefix(str(s).strip()))
    df["n_sources"] = pd.to_numeric(df["n_sources"], errors="coerce").fillna(0)
    df["chain_fit"] = pd.to_numeric(df["chain_fit"], errors="coerce").fillna(0)
    df = df.sort_values(["n_sources", "chain_fit"], ascending=False).drop_duplicates("k")
    out = {}
    for r in df.itertuples():
        out[r.k] = {"theme": str(r.target), "n_sources": int(r.n_sources),
                    "chain_fit": float(r.chain_fit),
                    "members_total": None if pd.isna(r.members_total) else int(r.members_total),
                    "moved": None if pd.isna(r.moved) else int(r.moved),
                    "mkt_trade_date": None if pd.isna(r.mkt_trade_date) else str(r.mkt_trade_date)}
    return out


def stock_daily(ds: str) -> dict[str, dict]:
    """数据日 ds 的个股行情三列：涨幅（百分数）、主力净额异常值、热度五日变化。"""
    try:
        df = db.read_pg(
            "SELECT ts_code, pct_change, net_z, heat_chg FROM v_factor_stock_daily "
            "WHERE trade_date = %s", (ds,))
    except Exception as e:  # noqa: BLE001
        print(f"  （个股日行情视图读取失败，候选卡的已启动门槛整体缺席: {e!r}）")
        return {}
    out = {}
    for r in df.itertuples():
        k = common.to_prefix(str(r.ts_code).strip())
        out[k] = {"pct0": _f(r.pct_change), "net_z": _f(r.net_z), "heat_chg": _f(r.heat_chg)}
    return out


def trading_days(upto: str, back_days: int = 90) -> list[str]:
    """交易日历：取一只长期存在的票在 gp_day_data 里的日期序列（表有一千四百万行，
    全表 DISTINCT 太慢；按 symbol 走索引）。返回升序 ISO 日期串。"""
    start = (dt.date.fromisoformat(upto) - dt.timedelta(days=back_days)).isoformat()
    try:
        df = db.read_mysql(
            "factor", "SELECT DISTINCT DATE(`timestamp`) d FROM gp_day_data "
                      "WHERE symbol = %s AND `timestamp` >= %s AND `timestamp` <= %s "
                      "ORDER BY d", ("SH600519", start, upto))
        return [pd.Timestamp(x).date().isoformat() for x in df["d"].tolist()]
    except Exception as e:  # noqa: BLE001
        print(f"  （交易日历读取失败，评分日龄按自然日×5/7 近似: {e!r}）")
        return []


def night_conclusions(codes, ds: str) -> dict[str, dict]:
    """决策系统昨夜结论（每票最新一行）：信号、支撑压力位、吸筹块与评分日龄。

    评分日龄 = 结论行的 trade_date 到数据日 ds 之间的交易日数（含头不含尾）。
    该表会被盘中补扫就地改写、没有落库时刻列，日龄以行的 trade_date 为准，
    这是已知局限（方案 2.2 第三项）。"""
    codes = sorted({common.to_prefix(str(c).strip()) for c in codes if c})
    if not codes:
        return {}
    # 只取不晚于数据日 ds 的结论行：生产上 ds 就是数据日，与"取最新一行"等价；
    # 复盘重建历史日时这一条防前视（否则历史日会读到今天的吸筹状态）。
    # trade_date 是整数 YYYYMMDD（数据源盘点 §1a），直接比大小。
    ds_int = int(ds.replace("-", "")) if ds else 99999999
    try:
        marks = ",".join(["%s"] * len(codes))
        df = db.read_mysql(
            "pms",
            f"SELECT stock_code, trade_date, signal_type, support_level, pressure_level, "
            f"raw_logic_json FROM strategy_daily_results "
            f"WHERE stock_code IN ({marks}) AND trade_date <= %s",
            tuple(codes) + (ds_int,))
    except Exception as e:  # noqa: BLE001
        print(f"  （决策系统结论表读取失败，吸筹确认线与坏信号风险整体缺席: {e!r}）")
        return {}
    if df.empty:
        return {}
    df = df.sort_values("trade_date").drop_duplicates("stock_code", keep="last")
    cal = trading_days(ds)
    cal_index = {d: i for i, d in enumerate(cal)}
    out = {}
    for r in df.itertuples():
        k = common.to_prefix(str(r.stock_code).strip())
        tdate = _ymd(r.trade_date)
        age = _age(tdate, ds, cal_index)
        ff = {}
        try:
            raw = json.loads(r.raw_logic_json) if isinstance(r.raw_logic_json, str) else (r.raw_logic_json or {})
            ff = (raw or {}).get("fund_flow") or {}
        except Exception:  # noqa: BLE001 —— 坏 JSON 当无吸筹块
            ff = {}
        out[k] = {
            "signal": str(r.signal_type or "").strip().upper() or None,
            "support": _f(r.support_level), "pressure": _f(r.pressure_level),
            "conclusion_date": tdate,
            "accum_state": str(ff.get("state") or "") or None,
            "accum_score": _f(ff.get("score")),
            "accum_structure": ff.get("structure"), "accum_pos_tag": ff.get("pos_tag"),
            "accum_age": age,
        }
    return out


def _ymd(v) -> str | None:
    """各表的日期列形态不一（整数 YYYYMMDD、date、datetime、ISO 串），统一成 ISO 日期串。
    前几种形态不经 pandas 就能认出来，这样离线单测里的假数据不依赖 pandas；认不出的最后才交给
    pandas 解析，仍失败返回 None。"""
    if v is None or (isinstance(v, float) and v != v):
        return None
    if isinstance(v, dt.datetime):
        return v.date().isoformat()
    if isinstance(v, dt.date):
        return v.isoformat()
    s = str(v).strip()
    if len(s) == 8 and s.isdigit():
        return f"{s[:4]}-{s[4:6]}-{s[6:]}"
    if len(s) >= 10 and s[4] == "-" and s[7] == "-" and s[:4].isdigit():
        return s[:10]
    try:
        return pd.Timestamp(s).date().isoformat()
    except Exception:  # noqa: BLE001
        return None


def _age(tdate: str | None, ds: str, cal_index: dict) -> int | None:
    if not tdate:
        return None
    if tdate in cal_index and ds in cal_index:
        return cal_index[ds] - cal_index[tdate]
    try:  # 日历缺失或日期在日历之外：自然日 × 5/7 近似
        nat = (dt.date.fromisoformat(ds) - dt.date.fromisoformat(tdate)).days
        return max(0, round(nat * 5 / 7))
    except ValueError:
        return None


def _f(v):
    try:
        x = float(v)
    except (TypeError, ValueError):
        return None
    return None if x != x else x


def _records(df) -> list[dict]:
    """把读函数的返回统一成字典列表：DataFrame 走 to_dict，普通列表原样返回，None 与空表返回空列表。
    这样取数函数的计算部分只面对普通 Python 对象，离线单测的假读函数直接返回字典列表即可。"""
    if df is None:
        return []
    if isinstance(df, list):
        return [dict(r) for r in df]
    if hasattr(df, "to_dict"):
        if getattr(df, "empty", False):
            return []
        return list(df.to_dict("records"))
    return list(df)


def _to_dot(code: str) -> str:
    """前缀式 SH600000 转成数据基座的点后缀式 600000.SH；已是点后缀式或纯数字则原样返回。"""
    s = str(code or "").strip().upper()
    if "." in s or len(s) < 3:
        return s
    if s[:2] in ("SH", "SZ", "BJ") and s[2:].isdigit():
        return f"{s[2:]}.{s[:2]}"
    return s


# ============================================================================
# 因果论断（数据基座 v_factor_logic，2026-09-03 方案第 3.3 节"候选卡读因果论断"）
# ============================================================================
_LOGIC_COLS = ("ts_code", "subject_name", "object_name", "direction", "mechanism", "condition",
               "horizon", "strength", "tier", "confidence", "disclosure_date", "doc_id",
               "doc_title", "source_span", "claim_id", "via_segment")


def logic_claims(codes, ds: str, per_stock: int | None = None, read_pg=None) -> dict[str, list[dict]]:
    """这批票在数据基座因果论断视图里、披露日不晚于数据日 ds 的论断，每票取最近披露日的最多
    per_stock 条（默认 config.LOGIC_CLAIMS_PER_STOCK），按前缀码索引。

    每条论断带：方向、机制、条件、时效、强度、层级、置信度、披露日、出处文档标题与编号、
    论断编号、经由环节、主体与客体名。只作展示与出处，不进判决；读失败返回空字典并打印一行原因。
    read_pg 可注入（离线单测），默认走 db.read_pg。"""
    n_per = config.LOGIC_CLAIMS_PER_STOCK if per_stock is None else int(per_stock)
    if n_per <= 0:
        return {}
    wanted = sorted({common.to_prefix(str(c).strip()) for c in codes if c})
    if not wanted:
        return {}
    dots = [_to_dot(c) for c in wanted]
    reader = read_pg or db.read_pg
    try:
        marks = ",".join(["%s"] * len(dots))
        rows = _records(reader(
            f"SELECT {', '.join(_LOGIC_COLS)} FROM v_factor_logic "
            f"WHERE ts_code IN ({marks}) AND disclosure_date <= %s",
            tuple(dots) + (ds,)))
    except Exception as e:  # noqa: BLE001
        print(f"  （因果论断视图 v_factor_logic 读取失败，候选卡的论断证据线整体缺席: {e!r}）")
        return {}
    if not rows:
        return {}
    want = set(wanted)
    by_code: dict[str, list[dict]] = {}
    seen: set[tuple[str, str]] = set()
    for r in rows:
        k = common.to_prefix(str(r.get("ts_code") or "").strip())
        if k not in want:                 # 只留请求的票，视图返回的多余行不带进卡
            continue
        cid = _s(r.get("claim_id"))
        if cid and (k, cid) in seen:      # 视图里客体为环节的论断按成员展开，同票同论断只留一条
            continue
        if cid:
            seen.add((k, cid))
        by_code.setdefault(k, []).append({
            "direction": _s(r.get("direction")), "mechanism": _s(r.get("mechanism")),
            "condition": _s(r.get("condition")), "horizon": _s(r.get("horizon")),
            "strength": _s(r.get("strength")), "tier": _s(r.get("tier")),
            "confidence": _f(r.get("confidence")),
            "disclosure_date": _ymd(r.get("disclosure_date")),
            "doc_id": _s(r.get("doc_id")), "doc_title": _s(r.get("doc_title")),
            "source_span": (_s(r.get("source_span")) or "")[:200] or None,
            "claim_id": _s(r.get("claim_id")), "via_segment": _s(r.get("via_segment")),
            "subject": _s(r.get("subject_name")), "object": _s(r.get("object_name")),
        })
    out = {}
    for k, items in by_code.items():
        items.sort(key=lambda c: (c["disclosure_date"] or "", c["confidence"] or -1.0), reverse=True)
        out[k] = items[:n_per]
    return out


def _s(v) -> str | None:
    if v is None or (isinstance(v, float) and v != v):
        return None
    s = str(v).strip()
    return s or None


# ============================================================================
# 计划环境段的市场四项（2026-09-03 方案第 1.4 节清单里"有"与"可自算"的项）
# ============================================================================
def market_context(ds: str, read_pg=None, read_mysql=None) -> dict:
    """数据日 ds 的市场环境四项，全部只展示与复盘分组，不拦任何票。

      turnover    两市成交额：指数日线表 zs_day_data 里上证与深成当日 amount 之和，以及相对前五个
                  交易日均值的比值（原表单位，未换算）
      breadth     市场广度：基座个股日行情视图当日行自算——上涨、下跌、平盘家数，涨幅达 9.8% 的家数
                  （涨停近似），涨幅中位数
      margin      融资：eastmoney_rzrq_data 最新一日的 financing_balance 与 change_percent_5d
      fear_greed  恐贪指数：fear_greed_index 最新一日的 index_value 与日期

    每一项读失败为 None 并把原因记进 errors，不阻断出计划。read_pg / read_mysql 可注入（离线单测）。
    融资与恐贪两张表按"最新一日"取，不按 ds 过滤：它们是 T+1 更新的情绪读数，计划日早晨看到的
    就是最新一行；行里带日期，读者自己判断新鲜度。"""
    rpg = read_pg or db.read_pg
    rmy = read_mysql or db.read_mysql
    src = config.MARKET_MYSQL_SOURCE
    out = {"date": ds, "turnover": None, "breadth": None, "margin": None, "fear_greed": None,
           "errors": {}, "fetched_at": dt.datetime.now().isoformat(timespec="seconds")}

    # 一、两市成交额。不按日期列过滤（列的类型未在本仓库内核实：DATE 与整数 YYYYMMDD 的比较
    # 语义不同），改为取每个指数最近的几十行，在 Python 里按归一化日期筛不晚于 ds 的行。
    try:
        marks = ",".join(["%s"] * len(MARKET_INDEX_CODES))
        rows = _records(rmy(
            src, f"SELECT symbol, `timestamp` AS d, amount FROM zs_day_data "
                 f"WHERE symbol IN ({marks}) ORDER BY `timestamp` DESC LIMIT 80",
            tuple(MARKET_INDEX_CODES)))
        out["turnover"] = _turnover(rows, ds)
        if out["turnover"] is None:
            out["errors"]["turnover"] = "zs_day_data 最近 40 个交易日内没有不晚于计划日、且两市齐全的行"
    except Exception as e:  # noqa: BLE001
        out["errors"]["turnover"] = repr(e)
        print(f"  （两市成交额读取失败，环境段该项为空: {e!r}）")

    # 二、市场广度：基座个股日行情视图当日全部行自算。
    try:
        rows = _records(rpg(
            "SELECT pct_change FROM v_factor_stock_daily WHERE trade_date = %s", (ds,)))
        out["breadth"] = _breadth(rows)
        if out["breadth"] is None:
            out["errors"]["breadth"] = "v_factor_stock_daily 当日无行"
    except Exception as e:  # noqa: BLE001
        out["errors"]["breadth"] = repr(e)
        print(f"  （市场广度自算失败，环境段该项为空: {e!r}）")

    # 三、融资余额与五日变化；四、恐贪指数。两张表都取最新一日一行。
    for key, table, cols, label in (
            ("margin", "eastmoney_rzrq_data", ("financing_balance", "change_percent_5d"), "融资余额"),
            ("fear_greed", "fear_greed_index", ("index_value",), "恐贪指数")):
        try:
            row, date_col = _latest_row(rmy, src, table)
            if row is None:
                out["errors"][key] = f"{table} 为空表"
                continue
            item = {"date": _ymd(row.get(date_col)) if date_col else None,
                    "date_col": date_col, "source": table}
            for c in cols:
                item[c] = _f(row.get(c))
            if all(item[c] is None for c in cols):
                out["errors"][key] = f"{table} 最新行缺列 {cols}（实际列: {sorted(row)[:12]}）"
                continue
            out[key] = item
        except Exception as e:  # noqa: BLE001
            out["errors"][key] = repr(e)
            print(f"  （{label}读取失败，环境段该项为空: {e!r}）")
    return out


def _turnover(rows: list[dict], ds: str) -> dict | None:
    """两市成交额：按日期把两个指数的 amount 相加，只认两市齐全的日子；当日取不晚于 ds 的最近一日，
    前五日均值取它之前的五个交易日（不足五个按实际个数）。"""
    by_day: dict[str, dict] = {}
    for r in rows:
        d = _ymd(r.get("d"))
        a = _f(r.get("amount"))
        if not d or a is None or d > ds:
            continue
        by_day.setdefault(d, {})[str(r.get("symbol") or "").strip()] = a
    full = sorted((d for d, m in by_day.items() if all(c in m for c in MARKET_INDEX_CODES)),
                  reverse=True)
    if not full:
        return None
    day0 = full[0]
    amt0 = sum(by_day[day0].values())
    prev = [sum(by_day[d].values()) for d in full[1:6]]
    avg5 = (sum(prev) / len(prev)) if prev else None
    return {"data_date": day0, "amount": amt0, "prev5_avg": avg5,
            "ratio_vs_prev5": (amt0 / avg5) if avg5 else None, "prev5_days": len(prev),
            "unit": "zs_day_data 原表单位，未换算",
            "source": "zs_day_data 上证 000001.SH 与深成 399001.SZ 当日 amount 之和"}


def _breadth(rows: list[dict]) -> dict | None:
    pcts = [p for p in (_f(r.get("pct_change")) for r in rows) if p is not None]
    if not pcts:
        return None
    return {"n": len(pcts),
            "up": sum(1 for p in pcts if p > 0), "down": sum(1 for p in pcts if p < 0),
            "flat": sum(1 for p in pcts if p == 0),
            "limit_up_approx": sum(1 for p in pcts if p >= LIMIT_UP_PCT),
            "pct_median": round(statistics.median(pcts), 3),
            "limit_up_rule": f"涨幅达 {LIMIT_UP_PCT}% 记为涨停近似",
            "source": "v_factor_stock_daily 当日行自算"}


def _latest_row(rmy, src: str, table: str) -> tuple[dict | None, str | None]:
    """取一张表按日期列排序的最新一行。日期列名先用一行样本探出（候选名见 _DATE_COL_CANDIDATES），
    探不到就按第一列倒序（通常是自增主键）并把 date_col 记为 None。"""
    sample = _records(rmy(src, f"SELECT * FROM {table} LIMIT 1"))
    if not sample:
        return None, None
    cols = list(sample[0].keys())
    date_col = next((c for c in _DATE_COL_CANDIDATES if c in cols), None)
    order = f"`{date_col}`" if date_col else "1"
    rows = _records(rmy(src, f"SELECT * FROM {table} ORDER BY {order} DESC LIMIT 1"))
    return (rows[0] if rows else None), date_col