akg-factor-bridge/test_market_context.py

"""sources.market_context 与 sources.logic_claims 的离线单测（不连库），另带复盘脚本的台账标题解析。

取数函数都接受注入的读函数：这里用返回字典列表的假函数替换 db.read_pg / db.read_mysql，
覆盖：两市成交额与前五日比值、日期筛选、两市不齐全的日子被跳过；广度四项；融资与恐贪按最新
一行取且日期列自动探测；每一项读失败为空不阻断；因果论断按前缀码索引、只取披露日不晚于
数据日的、每票最多三条按披露日倒序；读失败返回空字典。

开发机没有 pandas 与数据库驱动时，只给缺席的模块装最小桩（与 test_plan_verdict.py 同一约定：
仅在模块缺席时装桩，不覆盖真实模块），取数函数的计算部分不碰 pandas。

跑法：python3 test_market_context.py   或   pytest test_market_context.py
"""
import datetime as dt
import os
import sys
import types

_STUBS = ("pandas", "psycopg", "pymysql", "dotenv")
for _n in _STUBS:
    if _n not in sys.modules:
        try:
            __import__(_n)
        except ImportError:
            _m = types.ModuleType(_n)
            if _n == "pandas":                      # db.py / plan.py 的函数签名在定义时引用这两个名字
                _m.DataFrame = type("DataFrame", (), {})
                _m.Series = type("Series", (), {})
            sys.modules[_n] = _m

import config  # noqa: E402
import sources  # noqa: E402


def t(name, cond):
    assert cond, name
    print("  ok", name)


# ---------------------------------------------------------------- 假数据
DS = "2026-09-02"


def _zs_rows():
    """指数日线：六个交易日两市齐全，另有一天只有上证（该日应被跳过），还有一天晚于数据日。"""
    days = ["2026-08-25", "2026-08-26", "2026-08-27", "2026-08-28", "2026-08-31", "2026-09-01", "2026-09-02"]
    rows = []
    for i, d in enumerate(days):
        rows.append({"symbol": "000001.SH", "d": dt.date.fromisoformat(d), "amount": 6000.0 + i * 100})
        if d != "2026-08-27":                       # 这一天深成缺行
            rows.append({"symbol": "399001.SZ", "d": dt.date.fromisoformat(d), "amount": 8000.0 + i * 100})
    rows.append({"symbol": "000001.SH", "d": dt.date(2026, 9, 3), "amount": 99999.0})   # 晚于数据日
    rows.append({"symbol": "399001.SZ", "d": dt.date(2026, 9, 3), "amount": 99999.0})
    return rows


def _mysql_ok(src, sql, params=None):
    s = " ".join(sql.split())
    if "zs_day_data" in s:
        return _zs_rows()
    if "eastmoney_rzrq_data" in s:
        if "LIMIT 1" in s and "ORDER BY" not in s:
            return [{"id": 1, "stat_date": 20260901, "financing_balance": 1.9e12, "change_percent_5d": 1.23}]
        assert "ORDER BY `stat_date` DESC" in s, s
        return [{"id": 9, "stat_date": 20260901, "financing_balance": 1.9e12, "change_percent_5d": 1.23}]
    if "fear_greed_index" in s:
        if "LIMIT 1" in s and "ORDER BY" not in s:
            return [{"id": 1, "date": "2026-09-01", "index_value": 62.5}]
        assert "ORDER BY `date` DESC" in s, s
        return [{"id": 7, "date": "2026-09-01", "index_value": 62.5}]
    raise AssertionError(f"意外的查询: {s}")


def _pg_ok(sql, params=None):
    s = " ".join(sql.split())
    if "v_factor_stock_daily" in s:
        assert params == (DS,)
        return [{"pct_change": 9.95}, {"pct_change": 3.0}, {"pct_change": 0.0}, {"pct_change": -1.5},
                {"pct_change": None}, {"pct_change": 10.02}, {"pct_change": -4.0}]
    if "v_factor_logic" in s:
        assert params[-1] == DS and "600000.SH" in params and "SZ000001" not in params
        return [
            {"ts_code": "600000.SH", "direction": "利好", "mechanism": "机制甲", "condition": None, "horizon": "一年",
             "strength": "强", "tier": "T1", "confidence": 0.8, "disclosure_date": dt.date(2026, 8, 20),
             "doc_id": "d1", "doc_title": "文档一", "source_span": "x" * 300, "claim_id": "c1", "via_segment": "环节甲",
             "subject_name": "甲", "object_name": "乙"},
            {"ts_code": "600000.SH", "direction": "利好", "mechanism": "机制乙", "condition": "条件乙", "horizon": "半年",
             "strength": "中", "tier": "T2", "confidence": 0.6, "disclosure_date": "2026-08-30",
             "doc_id": "d2", "doc_title": "文档二", "source_span": None, "claim_id": "c2", "via_segment": None,
             "subject_name": "甲", "object_name": "丙"},
            {"ts_code": "600000.SH", "direction": "利空", "mechanism": "机制丙", "condition": None, "horizon": None,
             "strength": None, "tier": None, "confidence": 0.9, "disclosure_date": "2026-08-30",
             "doc_id": "d3", "doc_title": "文档三", "source_span": "", "claim_id": "c3", "via_segment": None,
             "subject_name": None, "object_name": None},
            {"ts_code": "600000.SH", "direction": "利好", "mechanism": "机制丁", "condition": None, "horizon": None,
             "strength": None, "tier": None, "confidence": 0.5, "disclosure_date": "2026-07-01",
             "doc_id": "d4", "doc_title": "文档四", "source_span": None, "claim_id": "c4", "via_segment": None,
             "subject_name": None, "object_name": None},
            {"ts_code": "000001.SZ", "direction": "利好", "mechanism": "机制戊", "condition": None, "horizon": None,
             "strength": None, "tier": None, "confidence": None, "disclosure_date": 20260815,
             "doc_id": "d5", "doc_title": "文档五", "source_span": None, "claim_id": "c5", "via_segment": None,
             "subject_name": None, "object_name": None},
        ]
    raise AssertionError(f"意外的查询: {s}")


def _boom(*a, **k):
    raise OSError("connection refused")


# ---------------------------------------------------------------- 用例
def test_market_context():
    config.MARKET_MYSQL_SOURCE = "price"
    m = sources.market_context(DS, read_pg=_pg_ok, read_mysql=_mysql_ok)
    tv = m["turnover"]
    t("两市成交额取数据日、两市齐全的行：6600+8600", tv and tv["data_date"] == DS and tv["amount"] == 15200.0)
    # 前五日：09-01(15000)、08-31(14800)、08-28(14600)、08-26(14200)；08-27 深成缺行被跳过 → 再补 08-25(14000)
    t("前五日均值跳过两市不齐全的日子", tv["prev5_days"] == 5 and abs(tv["prev5_avg"] - 14520.0) < 1e-6)
    t("比值 = 当日 / 前五日均值", abs(tv["ratio_vs_prev5"] - 15200.0 / 14520.0) < 1e-9)
    t("晚于数据日的行不参与", tv["amount"] < 99999)
    # 指数日线的 amount 单位是千元（2026-09-03 实测：09-02 两市合计 1,321,831,143，真实成交额约
    # 1.32 万亿元）。换算成亿元要除以十万。渲染只用 amount_yi，写错单位会让读数差一千倍。
    t("成交额单位标千元、按千元换算成亿元",
      tv["unit"] == "千元" and abs(tv["amount_yi"] - 15200.0 / 1e5) < 1e-12
      and abs(tv["prev5_avg_yi"] - 14520.0 / 1e5) < 1e-12)
    b = m["breadth"]
    # 六个有效值排序：-4.0、-1.5、0.0、3.0、9.95、10.02，中位数 = (0.0 + 3.0) / 2 = 1.5
    t("广度：上涨 3 / 下跌 2 / 平盘 1，涨停近似 2，中位数 1.5（空值剔除）",
      b["n"] == 6 and b["up"] == 3 and b["down"] == 2 and b["flat"] == 1 and b["limit_up_approx"] == 2
      and b["pct_median"] == 1.5)
    mg = m["margin"]
    t("融资：最新一行、日期列自动探到 stat_date、整数日期归一",
      mg and mg["date"] == "2026-09-01" and mg["date_col"] == "stat_date"
      and mg["financing_balance"] == 1.9e12 and mg["change_percent_5d"] == 1.23)
    fg = m["fear_greed"]
    t("恐贪：最新一行、日期列 date", fg and fg["index_value"] == 62.5 and fg["date"] == "2026-09-01")
    t("四项齐全时 errors 为空", m["errors"] == {} and m["date"] == DS)

    m = sources.market_context(DS, read_pg=_boom, read_mysql=_boom)
    t("四项读失败：全为空、原因入 errors、不抛错",
      m["turnover"] is None and m["breadth"] is None and m["margin"] is None and m["fear_greed"] is None
      and set(m["errors"]) == {"turnover", "breadth", "margin", "fear_greed"})

    def _mysql_partial(src, sql, params=None):
        if "zs_day_data" in sql:
            raise OSError("proxy down")
        return _mysql_ok(src, sql, params)
    m = sources.market_context(DS, read_pg=_pg_ok, read_mysql=_mysql_partial)
    t("单项失败不影响其余三项", m["turnover"] is None and "turnover" in m["errors"]
      and m["breadth"] and m["margin"] and m["fear_greed"])

    m = sources.market_context("2026-01-01", read_pg=lambda *a, **k: [], read_mysql=_mysql_ok)
    t("数据日早于所有行、广度无行：两项为空并注明", m["turnover"] is None and m["breadth"] is None
      and "turnover" in m["errors"] and "breadth" in m["errors"])


def test_logic_claims():
    config.LOGIC_CLAIMS_PER_STOCK = 3
    got = sources.logic_claims(["SH600000", "600000.SH", "SZ300750"], DS, read_pg=_pg_ok)
    t("按前缀码索引、去重后只查一次", set(got) == {"SH600000"})
    items = got["SH600000"]
    t("每票最多三条、按披露日倒序（同日按置信度）",
      [c["claim_id"] for c in items] == ["c3", "c2", "c1"])
    t("字段齐全：日期归一、出处、经由环节、原文截断到 200 字",
      items[2]["disclosure_date"] == "2026-08-20" and items[2]["doc_title"] == "文档一"
      and items[2]["via_segment"] == "环节甲" and len(items[2]["source_span"]) == 200
      and items[1]["source_span"] is None and items[0]["condition"] is None)
    got = sources.logic_claims(["SH600000"], DS, per_stock=1, read_pg=_pg_ok)
    t("条数上限参数生效", len(got["SH600000"]) == 1 and got["SH600000"][0]["claim_id"] == "c3")
    t("上限 0 = 不读视图", sources.logic_claims(["SH600000"], DS, per_stock=0, read_pg=_boom) == {})
    t("读失败返回空字典不抛错", sources.logic_claims(["SH600000"], DS, read_pg=_boom) == {})
    t("空代码集不查库", sources.logic_claims([], DS, read_pg=_boom) == {})
    t("整数日期也能归一", sources._ymd(20260815) == "2026-08-15" and sources._ymd("2026-08-15 10:00:00") == "2026-08-15")
    t("前缀式转点后缀式", sources._to_dot("SH600000") == "600000.SH" and sources._to_dot("600000.SH") == "600000.SH")


def test_decision_ledger_titles():
    import plan_review
    here = os.path.dirname(os.path.abspath(__file__))
    entries = plan_review.decision_ledger_entries(os.path.join(here, "docs", "复盘决定台账.md"))
    t("台账标题行解析出编号、日期、标题且含 013", entries and entries[0]["no"] == "001"
      and any(e["no"] == "013" and "关注" in e["title"] for e in entries)
      and all(len(e["date"]) == 10 for e in entries))
    t("台账文件缺失返回空列表", plan_review.decision_ledger_entries("/nonexistent/台账.md") == [])


def main():
    test_market_context()
    test_logic_claims()
    test_decision_ledger_titles()
    print("ALL OK — 市场四项 / 单项失败不阻断 / 因果论断索引与上限 / 台账标题解析 全部通过")


if __name__ == "__main__":
    main()