"""策略说明书生成器（app.quant.strategy_doc）测试 + 策略说明/公式/编辑接口测试。 目标（AGENT.md §24/§31）：说明书必须由 spec **真实推导**，并且 - 条件表达式的渲染必须与 `app.quant.selection._eval_condition` 的实际求值一致 （含 ref 语义、缺失值语义、in/not_in 的列表语义）； - 步骤必须与 `app.quant.local_engine` 的真实行为一致（择股日/调仓日/顺延/期末）； - 未知因子、未建模约束一律进 warnings（不编造、不假装支持）。 同一文件还覆盖策略 API 的说明接口与「原地更新（PUT）」，因为三者共用同一契约。 """ from __future__ import annotations from datetime import date import pandas as pd import pytest from app.api import deps from app.domain.entities.research import ( ConditionSpec, CostSpec, FactorSpec, ResearchSpec, SelectionSpec, UniverseSpec, ) from app.domain.entities.strategy import SelectionStrategy from app.infrastructure.persistence.sqlalchemy.base import Base from app.main import app from app.quant.factors import FactorDef from app.quant.selection import build_condition_fields, eligible_symbols from app.quant.strategy_doc import describe_strategy from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from conftest_quant import synthetic_daily def _spec(**kw) -> ResearchSpec: """高股息案例参数（n=30 → x=20、m=y=6、hfq、显式成本与最低佣金）。""" base = dict( type="backtest", universe=UniverseSpec(exclude_st=False, min_listing_days=250), price_adjustment="hfq", factors=[FactorSpec(name="dividend_yield", weight=1.0)], conditions=[ConditionSpec(field="dv_ratio", op="lte", value=30)], selection=SelectionSpec(top_n=30, hold_top_x=20, allow_substitute=False, defer_buy=True), rebalance="monthly", selection_interval_months=6, rebalance_interval_months=6, period=(date(2020, 1, 1), date(2024, 12, 31)), costs=CostSpec(min_commission=5.0), ) base.update(kw) return ResearchSpec(**base) class TestSummary: def test_summary_contains_factors_truncation_cycles_and_costs(self) -> None: doc = describe_strategy(_spec()) s = doc.summary assert "股息率" in s, "summary 必须由因子元数据推导出因子含义" assert "dividend_yield" in s assert "前 30 只候选池" in s and "前 20 只等权" in s, "必须说清两级截断 n → x" assert "每 6 个月重新择股" in s and "每 6 个月调仓" in s, "必须说清 m / y" assert "后复权" in s and "收盘价成交" in s assert "佣金 0.03%" in s and "印花税 0.05%" in s and "滑点 0.1%" in s def test_summary_lower_is_better_factor_方向(self) -> None: doc = describe_strategy(_spec(factors=[FactorSpec(name="volatility_20", weight=1.0)])) assert "从低到高排序" in doc.summary def test_summary_without_hold_top_x_uses_n(self) -> None: doc = describe_strategy( _spec(selection=SelectionSpec(top_n=20, allow_substitute=False), conditions=[]) ) assert "前 20 只等权持有" in doc.summary assert "先通过" not in doc.summary, "无 conditions 时不得声称有过滤条件" def test_summary_states_buy_fallback_mode(self) -> None: """买不进的处置决定实际持仓，必须出现在一句话说明里（三种模式各一）。""" defer = describe_strategy(_spec(conditions=[])) assert "买不进则顺延买入" in defer.summary subst = describe_strategy( _spec(selection=SelectionSpec(top_n=5, allow_substitute=True), conditions=[]) ) assert "从候选池之外替补" in subst.summary none = describe_strategy( _spec(selection=SelectionSpec(top_n=5, allow_substitute=False), conditions=[]) ) assert "买不进则放弃" in none.summary class TestFormula: def test_formula_has_factor_name_meaning_and_formula(self) -> None: doc = describe_strategy(_spec()) f = doc.formula assert "dividend_yield" in f # 因子元数据（AGENT.md §22）逐项落进公式：含义 / 公式 / 方向 / 输入列 assert "含义：股息率" in f assert "公式：dv_ratio（Tushare daily_basic，逐日时点值）" in f assert "越高越好" in f and "higher_is_better" in f assert "输入列：dv_ratio" in f assert "score_i = Σ_f w_f × d_f × z_f,i" in f assert "z_f,i = (x_f,i - mean_i(x_f)) / std_i(x_f)" in f def test_formula_has_two_level_truncation_and_cycles(self) -> None: f = describe_strategy(_spec()).formula assert "n = 30" in f and "x = 20" in f assert "每 6 个月" in f assert "择股日" in f and "调仓日" in f def test_formula_has_costs_and_execution_price_basis(self) -> None: f = describe_strategy(_spec()).formula assert "佣金 = max(投入资金 × 0.03%, 最低佣金 5 元/笔)" in f assert "印花税 0.05%（仅卖出）" in f, "印花税只对卖出计提（local_engine._rebalance）" assert "滑点 0.1%" in f assert "后复权" in f and "乘以 adjust_factor" not in f # 口径文案由 _ADJUST_TEXT 给出 assert "成交时点：调仓日收盘" in f assert "1,000,000 元" in f def test_formula_renders_conditions_readably(self) -> None: spec = _spec( conditions=[ ConditionSpec(field="dv_ratio", op="lte", value=30), ConditionSpec(field="close", op="gte", ref="ma60"), ConditionSpec(field="fundamental.roe", op="gte", value=15), ConditionSpec(field="static.industry", op="in", value=["白酒", "银行"]), ConditionSpec(field="static.industry", op="ne", value="白酒"), ConditionSpec(field="pe", op="lt", value=30.5), ] ) f = describe_strategy(spec).formula assert "dv_ratio <= 30" in f assert "close >= ma60" in f, "ref 条件必须渲染成「字段 op 字段」（同一股票同一日比较）" assert "fundamental.roe >= 15" in f assert 'static.industry ∈ ["白酒", "银行"]' in f assert 'static.industry != "白酒"' in f assert "pe < 30.5" in f # 说明字段域语义，避免使用者误以为 fundamental.* 是当日值 assert "announce_date <= 择股日" in f def test_empty_conditions_does_not_crash(self) -> None: doc = describe_strategy(_spec(conditions=[])) assert "（无）" in doc.formula assert any("未配置过滤条件" in w for w in doc.warnings) assert doc.steps def test_unknown_factor_goes_to_warnings_not_crash(self) -> None: doc = describe_strategy(_spec(factors=[FactorSpec(name="no_such_factor", weight=2.0)])) assert any("未知因子 no_such_factor" in w for w in doc.warnings) assert "元数据缺失" in doc.formula assert doc.summary # 仍然可用（说明不因未知因子而失败） def test_factor_meta_override_is_used(self) -> None: meta = { "my_signal": FactorDef( name="my_signal", description="自定义信号", formula="close / ma20", brief="实验因子", direction="lower_is_better", requires=("close",), ) } doc = describe_strategy( _spec(factors=[FactorSpec(name="my_signal", weight=0.5)], conditions=[]), factor_meta=meta, ) assert "自定义信号" in doc.formula and "公式：close / ma20" in doc.formula assert "越低越好" in doc.formula assert not any("未知因子" in w for w in doc.warnings) assert "从低到高排序" in doc.summary def test_unknown_condition_field_goes_to_warnings(self) -> None: doc = describe_strategy( _spec(conditions=[ConditionSpec(field="no_such_col", op="gt", value=1)]) ) assert any("条件字段 no_such_col 未识别" in w for w in doc.warnings) assert "no_such_col > 1" in doc.formula class TestConditionSemanticsMatchSelectionEvaluator: """渲染文本必须与 selection 的真实求值一致（以代码为准，不以注释/直觉为准）。""" def test_gte_ref_and_lte_value_match_eligible_symbols(self) -> None: syms = ["600000.SH", "600001.SH", "600002.SH"] daily = synthetic_daily({"600000.SH": 0.006, "600001.SH": -0.006, "600002.SH": 0.0005}, n=300) obs = pd.Timestamp("2024-12-31") conds = [ ConditionSpec(field="close", op="gte", ref="ma60"), ConditionSpec(field="momentum_20", op="lte", value=0.05), ] fields = build_condition_fields(daily, conds, obs) passed = eligible_symbols(sorted(syms), conds, {}, fields, {}) # 期望值直接按「渲染出的表达式」的语义计算，与求值器结果对齐 for sym in syms: expected = bool( fields["close"][sym] >= fields["ma60"][sym] and fields["momentum_20"][sym] <= 0.05 ) assert (sym in passed) == expected, f"{sym} 的渲染语义与求值器不一致" # 且文档里出现的就是这两条表达式 doc = describe_strategy(_spec(conditions=conds)) assert "close >= ma60" in doc.formula assert "momentum_20 <= 0.05" in doc.formula def test_in_not_in_and_missing_value_semantics(self) -> None: """in/not_in 的右操作数是列表，语义为 `left in right`；缺失值语义见下。""" conds = [ ConditionSpec(field="static.industry", op="in", value=["白酒"]), ConditionSpec(field="static.pe", op="ne", value=10), # 缺失字段 ] statics = {"A": {"industry": "白酒"}, "B": {"industry": "银行"}} passed = eligible_symbols(["A", "B"], conds, statics, {}, {}) # 代码事实（selection._compare）：op == "ne" 在 None 判定**之前**返回 left != right， # 因此字段缺失时 `ne` 判定为「不等于」→ 通过；其余运算符在缺失值上不通过。 assert set(passed) == {"A"} doc = describe_strategy(_spec(conditions=conds)) assert 'static.industry ∈ ["白酒"]' in doc.formula assert "static.pe != 10" in doc.formula assert "除 != 外一律判为「未通过」" in doc.formula def test_string_comparison_is_lexicographic_in_real_evaluator(self) -> None: """对照 selection._compare 的实际实现，修正其内联注释的说法。 `_compare` 的注释写「字符串会 ValueError → False」，但代码里 gt/gte/lt/lte 最终走 `_num_cmp(left, right, op)`，而 `_num_cmp` **不做 float 转换**， 因此两个字符串是按 Python 原生（字典序）比较的，并不会 ValueError。 说明书按代码事实渲染为 `static.industry > "M"`（同一运算符、同一语义）。 """ conds = [ConditionSpec(field="static.industry", op="gt", value="M")] statics = {"A": {"industry": "白酒"}} # '白酒' > 'M'（Unicode 码位更大） passed = eligible_symbols(["A"], conds, statics, {}, {}) assert set(passed) == {"A"} assert 'static.industry > "M"' in describe_strategy(_spec(conditions=conds)).formula class TestStepsAndWarnings: def test_steps_follow_local_engine_behaviour(self) -> None: doc = describe_strategy(_spec()) joined = " | ".join(doc.steps) assert "300 个自然日" in joined, "预热窗口来自 service._load_daily" assert "候选池" in joined and "selection_history" in joined assert "收盘" in joined, "成交发生在调仓日收盘" assert "涨停" in joined and "停牌" in joined assert "顺延" in joined, "defer_buy=True 必须如实说明顺延买入" assert "次日起计收益" in joined, "调仓日收盘生效（无未来函数）" assert "不强制平仓" in joined, "期末不平仓是引擎的真实行为，必须披露" def test_steps_describe_substitute_mode(self) -> None: doc = describe_strategy( _spec(selection=SelectionSpec(top_n=5, allow_substitute=True), conditions=[]) ) assert any("替补" in s for s in doc.steps) def test_selection_steps_carry_no_own_numbering(self) -> None: """选股策略的执行步骤不带自带序号：前端渲染进
    ，后端再写「1.」会双重编号。""" st = SelectionStrategy( name="高股息", description="x", factors=[FactorSpec(name="dividend_yield", weight=1)], conditions=[], ) doc = describe_strategy(st) assert doc.steps for s in doc.steps: assert not s.lstrip().startswith(("1.", "2.", "3.", "4.", "5.")), s # 步骤内容本身仍要在（去掉的只是序号） assert "股票池" in doc.steps[0] assert "因子" in doc.steps[-1] def test_condition_literals_carry_base_unit(self) -> None: """字面量条件必须带**基准单位**：库里存的就是基准单位值，裸数字会被读成别的量级。 单位只做界面换算（见 quant/condition_fields 的单位阶梯），所以文档里写「50000 万元」 才是引擎真正比较的口径；字段间比较（ref）两侧同单位，不加后缀。 """ st = SelectionStrategy( name="单位口径", description="x", factors=[FactorSpec(name="dividend_yield", weight=1)], conditions=[ ConditionSpec(field="total_mv", op="gte", value=50000), ConditionSpec(field="close", op="gt", ref="ma60"), ConditionSpec(field="industry", op="in", value=["银行", "白酒"]), ], ) doc = describe_strategy(st) joined = "\n".join([doc.formula, *doc.steps]) assert "total_mv >= 50000 万元" in joined assert "close > 字段 ma60" in joined and "ma60 元" not in joined assert "['银行', '白酒']" in joined def test_universe_step_conditional_and_time_accurate(self) -> None: """股票池步骤只写实际生效的过滤，且时点口径必须正确（起始日过滤一次）。 以 service._load_daily / universe.filter_stocks 的代码为准：universe 过滤在 as_of=回测起始日执行一次，之后只有 exclude_st 会按当日名称逐择股日重判。 """ doc = describe_strategy(_spec(universe=UniverseSpec(min_listing_days=250), conditions=[])) pool_step = doc.steps[1] assert "回测起始日按 universe 过滤一次" in pool_step assert "当日名称" in pool_step assert "指数成分" not in pool_step, "未配置 index_code 时不得声称有指数成分过滤" idx = describe_strategy( _spec( universe=UniverseSpec(index_code="000300.SH", exclude_st=False, min_listing_days=0), conditions=[], ) ) assert "000300.SH 的当日历史成分" in idx.steps[1] assert "名称含 ST" not in idx.steps[1] def test_warnings_cover_unmodelled_constraints(self) -> None: # exclude_st 默认 True（UniverseSpec 默认值）时才需要标注时点/快照口径 doc = describe_strategy(_spec(universe=UniverseSpec(min_listing_days=250))) joined = " | ".join(doc.warnings) assert "exclude_suspended" in joined, "停牌未建模必须标注" assert "名称变更历史" in joined, "exclude_st 时点/快照口径必须标注" assert "幸存者偏差" in joined assert "涨跌停" in joined def test_warnings_flag_none_adjustment_and_industry_cap(self) -> None: from app.domain.entities.research import PortfolioSpec doc = describe_strategy( _spec(price_adjustment="none", portfolio=PortfolioSpec(max_industry_weight_pct=0.3)) ) joined = " | ".join(doc.warnings) assert "不复权" in joined assert "未建模" in joined and "行业权重" in joined def test_stale_pool_warning_when_y_lt_m(self) -> None: doc = describe_strategy( _spec(selection_interval_months=12, rebalance_interval_months=3, conditions=[]) ) assert any("池子陈旧" in w for w in doc.warnings) class TestDefinitionInput: def test_selection_strategy_describes_only_selection(self) -> None: """选股策略说明只讲「怎么选」，不假装知道回测参数（重构后无 period/costs/selection）。""" st = SelectionStrategy( name="演示", factors=[FactorSpec(name="momentum_60", weight=1.0)], conditions=[{"field": "dv_ratio", "op": "lte", "value": 30}], ) doc = describe_strategy(st) assert doc.summary and doc.formula and doc.steps assert "选股策略" in doc.summary assert "momentum_60" in doc.formula # 如实声明回测参数不在策略内 assert any("回测组合" in w for w in doc.warnings) def test_real_spec_has_no_placeholder_warning(self) -> None: doc = describe_strategy(_spec()) assert not any("无回测区间" in w for w in doc.warnings) assert "2020-01-01 ~ 2024-12-31" in doc.formula def test_describe_does_not_mutate_input(self) -> None: """纯函数约定：只生成文本，不改写 spec（更不改写研究结果）。""" spec = _spec() before = spec.model_dump() describe_strategy(spec) assert spec.model_dump() == before st = SelectionStrategy(name="演示", factors=[FactorSpec(name="momentum_60")]) before_st = st.model_dump() describe_strategy(st) assert st.model_dump() == before_st def test_unsupported_input_raises_clear_type_error(self) -> None: with pytest.raises(TypeError, match="ResearchSpec 或 SelectionStrategy"): describe_strategy({"factors": []}) # type: ignore[arg-type] # ---------- 策略 API：说明接口 + 原地更新（PUT） ---------- @pytest.fixture() def client(tmp_path) -> TestClient: engine = create_engine(f"sqlite:///{tmp_path / 'stratdoc.db'}", future=True) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine, expire_on_commit=False) def _session_override(): with Session() as s: yield s app.dependency_overrides[deps.get_session] = _session_override with TestClient(app) as c: yield c app.dependency_overrides.clear() _SPEC_BODY = { "type": "backtest", "universe": {"exclude_st": False, "min_listing_days": 0}, "factors": [{"name": "dividend_yield", "weight": 1.0}], "selection": {"top_n": 20, "hold_top_x": 20, "allow_substitute": False, "defer_buy": True}, "rebalance": "monthly", "selection_interval_months": 6, "rebalance_interval_months": 6, "period": ["2020-01-01", "2024-12-31"], } class TestStrategyDocApi: def test_post_describe_from_research_spec_without_saving(self, client: TestClient) -> None: resp = client.post("/api/strategies/describe", json=_SPEC_BODY) assert resp.status_code == 200 doc = resp.json() assert set(doc) == {"summary", "formula", "steps", "warnings"} assert "dividend_yield" in doc["formula"] assert "每 6 个月" in doc["summary"] # 未保存：列表仍为空（POST /describe 与 POST /strategies 不冲突） assert client.get("/api/strategies").json() == [] def test_post_describe_invalid_spec_422(self, client: TestClient) -> None: bad = dict(_SPEC_BODY, period=["2024-01-01", "2020-01-01"]) assert client.post("/api/strategies/describe", json=bad).status_code == 422 def test_get_describe_saved_strategy_and_404(self, client: TestClient) -> None: created = client.post( "/api/strategies", json={"name": "高股息", "factors": [{"name": "dividend_yield", "weight": 1}]}, ) sid = created.json()["id"] resp = client.get(f"/api/strategies/{sid}/describe") assert resp.status_code == 200 doc = resp.json() assert doc["summary"] assert "选股策略" in doc["summary"] assert client.get("/api/strategies/STG-NOT-EXIST/describe").status_code == 404 def test_post_fills_empty_description(self, client: TestClient) -> None: """需求：策略必须有说明 —— description 为空/纯空白时由 summary 自动补全。""" for desc in ("", " "): resp = client.post( "/api/strategies", json={ "name": f"无说明策略-{len(desc)}", "description": desc, "factors": [{"name": "momentum_60", "weight": 1}], }, ) assert resp.status_code == 200 body = resp.json() assert body["description"].strip() assert "momentum_60" in body["description"] # 响应中的说明与 GET 读回一致（真的落库了） assert client.get(f"/api/strategies/{body['id']}").json()["description"] == body[ "description" ] def test_post_keeps_explicit_description(self, client: TestClient) -> None: resp = client.post( "/api/strategies", json={ "name": "有说明策略", "description": "我自己写的说明", "factors": [{"name": "momentum_60", "weight": 1}], }, ) assert resp.json()["description"] == "我自己写的说明" def test_auto_description_fits_column_width(self, client: TestClient) -> None: """自动说明必须落在 `StrategyModel.description = String(300)` 之内。 重构后选股策略的说明只讲「怎么选」，天然简洁（不再拼回测公式）， 即使挂满全部因子也远低于列宽 —— 这里断言「一定放得下」即可； 截断分支（超长带省略号）由 ResearchSpec 路径保留，选股策略触达不到。 """ from app.api.strategies import _DESCRIPTION_MAX_CHARS from app.quant.factors import list_factors all_factors = [{"name": f.name, "weight": 1} for f in list_factors()] resp = client.post( "/api/strategies", json={"name": "全因子策略", "factors": all_factors} ) assert resp.status_code == 200 desc = resp.json()["description"] assert len(desc) <= _DESCRIPTION_MAX_CHARS, "选股策略说明也必须落在列宽内" normal = client.post( "/api/strategies", json={"name": "单因子策略", "factors": [{"name": "dividend_yield", "weight": 1}]}, ).json()["description"] assert len(normal) <= _DESCRIPTION_MAX_CHARS assert not normal.endswith("…") class TestStrategyUpdateApi: def _create(self, client: TestClient, name: str = "策略A") -> dict: resp = client.post( "/api/strategies", json={ "name": name, "description": "初始说明", "factors": [{"name": "momentum_60", "weight": 1}], }, ) assert resp.status_code == 200 return resp.json() def test_put_updates_in_place_and_keeps_created_at(self, client: TestClient) -> None: created = self._create(client) sid = created["id"] created_at = created["created_at"] assert created_at is not None resp = client.put( f"/api/strategies/{sid}", json={ "name": "策略A", "description": "改后的说明", "factors": [{"name": "momentum_20", "weight": 2}], }, ) assert resp.status_code == 200 body = resp.json() assert body["id"] == sid, "原地更新必须保持 id 不变（不新建）" assert body["created_at"] == created_at, "PUT 不得刷新创建时间" assert body["factors"][0]["name"] == "momentum_20" assert body["description"] == "改后的说明" # 再读一次：落库后的创建时间同样未变，且列表仍只有一条 assert client.get(f"/api/strategies/{sid}").json()["created_at"] == created_at assert len(client.get("/api/strategies").json()) == 1 def test_put_path_id_wins_over_body_id(self, client: TestClient) -> None: created = self._create(client, "策略B") sid = created["id"] resp = client.put( f"/api/strategies/{sid}", json={ "id": "STG-FAKE", "name": "策略B", "factors": [{"name": "momentum_60", "weight": 1}], }, ) assert resp.status_code == 200 assert resp.json()["id"] == sid assert client.get("/api/strategies/STG-FAKE").status_code == 404 def test_put_missing_strategy_404(self, client: TestClient) -> None: resp = client.put( "/api/strategies/STG-NOT-EXIST", json={"name": "X", "factors": [{"name": "momentum_60", "weight": 1}]}, ) assert resp.status_code == 404 assert "不存在" in resp.json()["detail"] def test_put_duplicate_name_400(self, client: TestClient) -> None: first = self._create(client, "策略C") second = self._create(client, "策略D") resp = client.put( f"/api/strategies/{second['id']}", json={"name": "策略C", "factors": [{"name": "momentum_60", "weight": 1}]}, ) assert resp.status_code == 400 assert "策略名已存在" in resp.json()["detail"] # 撞车失败后第二条策略名不变（未被部分写入） assert client.get(f"/api/strategies/{second['id']}").json()["name"] == "策略D" assert client.get(f"/api/strategies/{first['id']}").json()["name"] == "策略C" def test_put_fills_empty_description(self, client: TestClient) -> None: created = self._create(client, "策略E") resp = client.put( f"/api/strategies/{created['id']}", json={ "name": "策略E", "description": " ", "factors": [{"name": "dividend_yield", "weight": 1}], }, ) assert resp.status_code == 200 assert resp.json()["description"].strip() assert "dividend_yield" in resp.json()["description"]