"""Experiment 归档补齐（2026-09）测试。 覆盖五件事： 1. **同步端点归档**：`POST /api/backtests` / `POST /api/factor-tests` 落库（此前只写 进程内存），并用响应头 `X-Experiment-Id` 返回归档 id；归档失败时仍返回计算结果 并用 `X-Archive-Error` 如实暴露（AGENT §7 不静默）。 2. **data_version**：真实数据指纹（交易日 + 各表行数），格式与列宽（varchar(40)）合规。 3. **列表过滤/分页**：`kind` / `q` / `limit` / `offset` + `X-Total-Count`（不静默截断）。 4. **DELETE**：200 → 再 GET 404 → 再 DELETE 404，且不触碰 job 记录。 5. **归档完整性**：默认不再按 60 只截断个股曲线；超体积预算时裁剪并留下 `archive_meta`（机器可读）+ `unimplemented`（人可读）证据； 完整结果只存 experiment 一份（`GET /api/jobs/{id}` 契约不变，老记录可回退）。 """ from __future__ import annotations import json import re from datetime import date, datetime from types import SimpleNamespace import pytest from app.api import deps from app.application.services import experiment_archive as ea from app.domain.entities.market import Stock from app.domain.entities.research import ( BacktestResult, CostSpec, ExperimentRecord, ExperimentSummary, FactorSpec, JobRecord, JobStatus, ResearchSpec, SelectionSpec, UniverseSpec, ) from app.infrastructure.persistence.sqlalchemy.base import Base from app.infrastructure.persistence.sqlalchemy.models.jobs import ExperimentModel, JobModel from app.infrastructure.persistence.sqlalchemy.models.market import StockDailyModel from app.infrastructure.persistence.sqlalchemy.repositories.jobs_impl import ( SqlAlchemyExperimentRepository, SqlAlchemyJobRepository, ) from app.main import app from app.quant.engine import LocalEngine from app.quant.service import ResearchService from fastapi.testclient import TestClient from sqlalchemy import create_engine, select from sqlalchemy.orm import sessionmaker from conftest_quant import bars_dataframe_to_daily_bars, synthetic_daily _SYMS = ["60000" + str(i) + ".SH" for i in range(5)] _BACKTEST_BODY = { "type": "backtest", "universe": {"exclude_st": False, "min_listing_days": 0}, "factors": [{"name": "momentum_20", "weight": 1.0}], "selection": {"top_n": 2}, "rebalance": "monthly", "period": ["2024-03-01", "2024-10-31"], } # data_version 格式：d，后跟可选的行数段（MySQL 近似值带 ≈ 与 k） _DATA_VERSION_RE = re.compile( r"^d(\d{8}|-)(;n(≈\d+k|\d+))?(;a(≈\d+k|\d+))?(;b(≈\d+k|\d+))?$" ) class _MemStockRepo: def __init__(self, stocks: list[Stock]) -> None: self._stocks = stocks def get_by_symbol(self, symbol: str) -> Stock | None: return next((s for s in self._stocks if s.symbol == symbol), None) def list(self) -> list[Stock]: return self._stocks class _MemDailyRepo: def __init__(self, bars) -> None: self._bars = bars def get_range_many(self, symbols, start, end, adjust="none"): return [b for b in self._bars if b.symbol in symbols and start <= b.trade_date <= end] def get_range(self, symbol, start, end): return [b for b in self._bars if b.symbol == symbol and start <= b.trade_date <= end] @pytest.fixture() def api(tmp_path): """TestClient + 临时 SQLite（job/experiment/行情表）+ 内存行情仓储。""" drifts = {sym: 0.003 - 0.0015 * i for i, sym in enumerate(_SYMS)} bars = bars_dataframe_to_daily_bars(synthetic_daily(drifts, n=300)) stocks = [ Stock(symbol=sym, name=f"测试股份{i}", list_date=date(1999, 11, 10)) for i, sym in enumerate(_SYMS) ] service = ResearchService(_MemStockRepo(stocks), _MemDailyRepo(bars), LocalEngine()) engine = create_engine(f"sqlite:///{tmp_path / 'archive.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 app.dependency_overrides[deps._service_factory] = lambda: service app.dependency_overrides[deps._stock_repo_factory] = lambda: _MemStockRepo(stocks) with TestClient(app) as client: yield client, Session app.dependency_overrides.clear() def _exp_by_id(Session, exp_id: str) -> ExperimentRecord | None: with Session() as session: return SqlAlchemyExperimentRepository(session).get(exp_id) class TestSyncEndpointsArchive: """任务 1：同步端点必须落库，并通过 X-Experiment-Id 暴露归档 id。""" def test_backtest_archives_to_db_and_returns_header(self, api) -> None: client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) assert resp.status_code == 200 body = resp.json() assert body["equity_curve"], "body 形状不变：仍是完整 BacktestResult" exp_id = resp.headers.get("X-Experiment-Id") assert exp_id and exp_id.startswith("EXP-"), "必须用响应头返回归档 id" exp = _exp_by_id(Session, exp_id) assert exp is not None, "同步回测必须在 experiment 表落库（不再只写进程内存）" assert exp.kind == "backtest" assert json.loads(exp.spec_json)["factors"][0]["name"] == "momentum_20" assert exp.summary_text and "总收益" in exp.summary_text assert exp.code_version, "归档必须带代码版本" assert exp.job_id is None, "同步端点没有 Job，job_id 应为 None" # 归档结果可解码为完整 BacktestResult archived = BacktestResult.model_validate_json(exp.result_json) assert archived.summary.total_return_pct == body["summary"]["total_return_pct"] # 详情接口能读回 detail = client.get(f"/api/experiments/{exp_id}") assert detail.status_code == 200 assert detail.json()["result"]["summary"]["total_return_pct"] == ( body["summary"]["total_return_pct"] ) def test_factor_test_archives_with_kind(self, api) -> None: client, Session = api payload = dict(_BACKTEST_BODY, type="factor_test") resp = client.post("/api/factor-tests", json=payload) assert resp.status_code == 200 assert resp.json()["factor_name"] == "momentum_20" exp_id = resp.headers.get("X-Experiment-Id") assert exp_id exp = _exp_by_id(Session, exp_id) assert exp is not None assert exp.kind == "factor_test" assert exp.result_json and json.loads(exp.result_json)["factor_name"] == "momentum_20" def test_archive_failure_keeps_result_and_exposes_reason(self, api, monkeypatch) -> None: """归档失败不得吞掉计算结果，且失败必须如实暴露（AGENT §7 / §24）。""" client, _Session = api def _boom(self, experiment): # noqa: ANN001 raise RuntimeError("db down") monkeypatch.setattr(SqlAlchemyExperimentRepository, "save", _boom) resp = client.post("/api/backtests", json=_BACKTEST_BODY) assert resp.status_code == 200, "归档失败也必须返回已算出的结果" assert resp.json()["equity_curve"], "结果本身必须完整返回" assert "X-Experiment-Id" not in resp.headers err = resp.headers.get("X-Archive-Error") assert err and "RuntimeError" in err and "db down" in err, ( f"归档失败原因必须如实暴露，实际：{err!r}" ) class TestDataVersion: """任务 2：data_version 必须是真实、可比较、且塞得进 varchar(40) 的指纹。""" def test_format_and_column_width(self, api) -> None: client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"]) assert exp is not None assert exp.data_version, "data_version 不允许为空（此前永远是 NULL）" assert _DATA_VERSION_RE.match(exp.data_version), exp.data_version assert len(exp.data_version) <= 40, "experiment.data_version 是 varchar(40)" def test_real_values_from_seeded_bars(self, api) -> None: """种子 3 行日线（SQLite 走精确 COUNT(*)）→ 指纹必须是精确的真实值。""" client, Session = api days = [date(2024, 9, 26), date(2024, 9, 27), date(2024, 9, 30)] with Session() as session: session.execute( StockDailyModel.__table__.insert(), [ { "symbol": f"60010{i}.SH", "trade_date": d, "source": "tushare", "adjust": "none", "open": 1.0, "high": 1.0, "low": 1.0, "close": 1.0, "volume": 1.0, "amount": 1.0, } for i, d in enumerate(days) ], ) session.commit() resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"]) assert exp is not None assert exp.data_version == "d20240930;n3;a0;b0", exp.data_version def test_unavailable_when_tables_missing(self, tmp_path) -> None: """取不到数据口径时如实降级为 unavailable，绝不编造数字。""" from app.infrastructure.persistence.sqlalchemy.data_version import compute_data_version engine = create_engine(f"sqlite:///{tmp_path / 'empty.db'}", future=True) with sessionmaker(bind=engine)() as session: assert compute_data_version(session) == "unavailable" class TestListFilters: """任务 3：列表过滤/分页在 SQL 层完成，并用 X-Total-Count 暴露总数。""" @staticmethod def _seed(Session) -> list[str]: """直接造 4 条归档：2 个 backtest（其一含 DIVIDEND_YIELD 因子）、1 个 factor_test、1 个 selection。""" rows = [ ("EXP-A1", "backtest", "DIVIDEND_YIELD", "总收益 12.00% · 年化 8.00% · 回撤 5.00%"), ("EXP-A2", "backtest", "momentum_20", "总收益 3.00% · 年化 2.00% · 回撤 1.00%"), ("EXP-A3", "factor_test", "momentum_20", "IC 0.0100 · RankIC 0.0200 · 样本 100 日"), ("EXP-A4", "selection", "momentum_60", "as_of 2024-10-31 · 选出 3 / 评估 100"), ] with Session() as session: repo = SqlAlchemyExperimentRepository(session) for i, (exp_id, kind, factor, summary) in enumerate(rows): repo.save( ExperimentRecord( id=exp_id, kind=kind, spec_json=json.dumps( { "factors": [{"name": factor, "weight": 1.0}], "period": ["2024-03-01", "2024-10-31"], "rebalance": "monthly", "selection": {"top_n": i + 1}, } ), result_json=json.dumps({"kind": kind, "n": i}), summary_text=summary, code_version="abc1234", data_version="d20260904;n1;a2;b3", created_at=datetime(2026, 9, 4, 10, i, 0), ) ) session.commit() return [r[0] for r in rows] def test_kind_filter_and_total_count(self, api) -> None: client, Session = api self._seed(Session) resp = client.get("/api/experiments", params={"kind": "backtest"}) assert resp.status_code == 200 body = resp.json() assert {e["id"] for e in body} == {"EXP-A1", "EXP-A2"} assert resp.headers["X-Total-Count"] == "2" def test_q_is_case_insensitive_over_factor_name_and_summary(self, api) -> None: client, Session = api self._seed(Session) # 因子名（在 spec_json 里，大小写不敏感） r1 = client.get("/api/experiments", params={"q": "dividend_yield"}) assert [e["id"] for e in r1.json()] == ["EXP-A1"] assert r1.headers["X-Total-Count"] == "1" # id 片段 r2 = client.get("/api/experiments", params={"q": "exp-a"}) assert r2.headers["X-Total-Count"] == "4" # summary_text（中文） r3 = client.get("/api/experiments", params={"q": "选出 3"}) assert [e["id"] for e in r3.json()] == ["EXP-A4"] # kind + q 组合 r4 = client.get("/api/experiments", params={"kind": "backtest", "q": "momentum"}) assert [e["id"] for e in r4.json()] == ["EXP-A2"] def test_like_wildcards_and_injection_are_bound(self, api) -> None: client, Session = api self._seed(Session) # `%` 是 LIKE 通配符，必须被转义成字面量：种子里只有 EXP-A1/A2 的摘要带 # 百分号，若未转义会返回全部 4 条 r = client.get("/api/experiments", params={"q": "%"}) assert {e["id"] for e in r.json()} == {"EXP-A1", "EXP-A2"} assert r.headers["X-Total-Count"] == "2" # 注入尝试必须是普通字符串（参数绑定），不得改变语义 inj = client.get("/api/experiments", params={"q": "' OR 1=1 --"}) assert inj.json() == [] and inj.headers["X-Total-Count"] == "0" def test_limit_offset_and_total_count_not_silently_truncated(self, api) -> None: client, Session = api self._seed(Session) page1 = client.get("/api/experiments", params={"limit": 2, "offset": 0}) assert len(page1.json()) == 2 assert page1.headers["X-Total-Count"] == "4", "总数必须暴露，客户端才知道被截断" page2 = client.get("/api/experiments", params={"limit": 2, "offset": 2}) assert len(page2.json()) == 2 ids = {e["id"] for e in page1.json()} | {e["id"] for e in page2.json()} assert len(ids) == 4, "分页不得重复/漏项（created_at 同秒时靠 id 次级排序稳定）" # 默认 limit（200）远大于旧硬编码的 50 assert client.get("/api/experiments").status_code == 200 def test_meta_includes_data_version_job_id_and_result_bytes(self, api) -> None: client, Session = api seeded = self._seed(Session) row = next(e for e in client.get("/api/experiments").json() if e["id"] == "EXP-A1") assert row["data_version"] == "d20260904;n1;a2;b3", "列表必须暴露数据指纹" assert row["job_id"] is None assert row["result_bytes"] == len(json.dumps({"kind": "backtest", "n": 0})) # 详情路径（result_json 在内存，len() 零成本）与列表口径必须一致 resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp_id = resp.headers["X-Experiment-Id"] exp = _exp_by_id(Session, exp_id) assert exp is not None detail = client.get(f"/api/experiments/{exp_id}").json() assert detail["result_bytes"] == len(exp.result_json) assert detail["data_version"] == exp.data_version assert detail["job_id"] is None listed = next(e for e in client.get("/api/experiments").json() if e["id"] == exp_id) assert listed["result_bytes"] == detail["result_bytes"], "列表与详情体积口径一致" assert set(seeded) == {"EXP-A1", "EXP-A2", "EXP-A3", "EXP-A4"} def test_repository_list_filtered_returns_summaries_without_result_json(self, api) -> None: _client, Session = api self._seed(Session) with Session() as session: repo = SqlAlchemyExperimentRepository(session) rows = repo.list_filtered(kind="backtest", limit=10) assert all(isinstance(r, ExperimentSummary) for r in rows) assert not hasattr(rows[0], "result_json"), "列表不得拉取 MEDIUMTEXT 大字段" assert repo.count_filtered() == 4 assert repo.count_filtered(kind="selection") == 1 class TestDelete: """任务 4：DELETE 只删归档，不删 Job 记录。""" def test_delete_then_404(self, api) -> None: client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp_id = resp.headers["X-Experiment-Id"] deleted = client.delete(f"/api/experiments/{exp_id}") assert deleted.status_code == 200 assert deleted.json() == {"deleted": exp_id} assert _exp_by_id(Session, exp_id) is None assert client.get(f"/api/experiments/{exp_id}").status_code == 404 again = client.delete(f"/api/experiments/{exp_id}") assert again.status_code == 404 assert "不存在" in again.json()["detail"] def test_delete_keeps_job_record_and_reports_missing_result(self, api) -> None: """删除归档不触碰 job 表；job 的结果只剩归档一份，故如实说明原因而非静默 200 空值。""" client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp_id = resp.headers["X-Experiment-Id"] with Session() as session: job = JobRecord( id="JOB-DEL-1", kind="backtest", spec_json=json.dumps(_BACKTEST_BODY), status=JobStatus.SUCCESS, experiment_id=exp_id, created_at=datetime.now(), ) SqlAlchemyJobRepository(session).create(job) session.commit() assert client.delete(f"/api/experiments/{exp_id}").status_code == 200 with Session() as session: kept = SqlAlchemyJobRepository(session).get("JOB-DEL-1") assert kept is not None, "job 记录是执行历史，DELETE 不得连带删除" assert kept.experiment_id == exp_id view = client.get("/api/jobs/JOB-DEL-1") assert view.status_code == 200 assert view.json()["result"] is None assert "已不存在" in view.json()["result_unavailable_reason"] class TestCurveCompleteness: """P0：默认完整存档（不再按 60 只截断）；超预算时裁剪并留证据。""" @staticmethod def _spec(top_n: int, symbols: list[str]) -> ResearchSpec: return ResearchSpec( type="backtest", universe=UniverseSpec( exclude_st=False, min_listing_days=0, symbols=list(symbols) ), factors=[FactorSpec(name="momentum_20")], selection=SelectionSpec(top_n=top_n, hold_top_x=top_n, allow_substitute=False, defer_buy=True), rebalance="monthly", selection_interval_months=6, rebalance_interval_months=6, period=(date(2024, 3, 1), date(2024, 6, 30)), costs=CostSpec( commission_rate=0.0, stamp_tax_rate=0.0, slippage_rate=0.0 ), ) def test_default_stores_every_held_symbol_beyond_60(self) -> None: syms = [f"60{i:04d}.SH" for i in range(70)] drifts = {s: 0.001 * (i % 4) + 0.0005 for i, s in enumerate(syms)} daily = synthetic_daily(drifts, n=120) res = LocalEngine().run_backtest(daily, self._spec(len(syms), syms)) held = {a.symbol for a in res.fills if a.symbol} assert len(held) > 60, f"构造的样本应持有 60 只以上，实际 {len(held)}" curve_syms = {c.symbol for c in res.symbol_curves} assert held <= curve_syms, "期内持有的每只都必须有曲线（完整存档，默认不截断）" assert not [n for n in res.unimplemented if "个股收益曲线" in n], "默认不得出现截断说明" def test_configured_limit_truncates_and_notes_actual_limit( self, monkeypatch: pytest.MonkeyPatch ) -> None: """配置了上限就截断 —— 且 note 里写清「实际保留几只 / 共几只」。""" from app.core import config as config_mod from app.quant import local_engine as le monkeypatch.setattr( config_mod, "get_settings", lambda: SimpleNamespace(research_archive_curve_limit=3), ) assert le._MAX_SYMBOL_CURVES is None, "默认代码路径不设上限（配置优先）" syms = [f"60{i:04d}.SH" for i in range(10)] drifts = {s: 0.001 * (i % 3) + 0.0005 for i, s in enumerate(syms)} daily = synthetic_daily(drifts, n=90) res = LocalEngine().run_backtest(daily, self._spec(len(syms), syms)) assert len(res.symbol_curves) == 3 note = [n for n in res.unimplemented if "个股收益曲线" in n] assert note and "3 只" in note[0] and "共持有" in note[0] def test_archive_budget_truncates_with_machine_readable_evidence( self, api, monkeypatch ) -> None: """归档字节预算：超预算时裁剪曲线，并把证据同时写进 archive_meta 与 unimplemented。 预算不靠猜：先按默认预算归档一次拿到「完整归档的字节数」L，再把预算设为 L - 2,000 重新归档 —— 必然需要裁掉若干条曲线，且裁剪后必须真的落到预算内。 """ client, Session = api body = { **_BACKTEST_BODY, "selection": {"top_n": 5}, "universe": {**_BACKTEST_BODY["universe"], "symbols": _SYMS}, } first = client.post("/api/backtests", json=body) assert first.status_code == 200 assert first.json()["archive_meta"]["truncated"] is False, "默认预算下应完整存档" full = _exp_by_id(Session, first.headers["X-Experiment-Id"]) assert full is not None full_bytes = len(full.result_json.encode("utf-8")) assert first.json()["archive_meta"]["curves_total"] >= 3, "样本需有足够曲线才能验证裁剪" budget = full_bytes - 2_000 monkeypatch.setattr(ea, "_archive_budget_bytes", lambda: budget) resp = client.post("/api/backtests", json=body) assert resp.status_code == 200 meta = resp.json()["archive_meta"] assert meta["truncated"] is True, f"预算 {budget} < 完整体积 {full_bytes}，必须触发裁剪" assert 0 < meta["curves_stored"] < meta["curves_total"], meta assert meta["budget_bytes"] == budget assert meta["over_budget"] is False assert any("个股收益曲线" in n for n in resp.json()["unimplemented"]) exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"]) assert exp is not None assert len(exp.result_json.encode("utf-8")) <= budget, "裁剪后必须真的放进预算" archived = json.loads(exp.result_json) assert archived["archive_meta"]["truncated"] is True assert archived["archive_meta"]["curves_stored"] == len(archived["symbol_curves"]) assert any("个股收益曲线" in n for n in archived["unimplemented"]) assert archived["archive_meta"]["result_bytes"] == len(exp.result_json.encode("utf-8")) def test_archive_over_budget_is_labelled_not_hidden(self, api, monkeypatch) -> None: """预算小到连归档主体都放不下：必须标 over_budget + 说明，绝不假装达标。""" client, Session = api budget = 5_000 monkeypatch.setattr(ea, "_archive_budget_bytes", lambda: budget) resp = client.post("/api/backtests", json=_BACKTEST_BODY) assert resp.status_code == 200 body = resp.json() meta = body["archive_meta"] assert meta["curves_stored"] == 0 and meta["truncated"] is True assert meta["over_budget"] is True, "主体超预算必须显式标注，而不是静默超标" assert any("超过配置的体积预算" in n for n in body["unimplemented"]) exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"]) assert exp is not None archived = json.loads(exp.result_json) assert archived["archive_meta"]["over_budget"] is True assert any("超过配置的体积预算" in n for n in archived["unimplemented"]) # 如实记录「确实超标」这一事实（不掩盖，也不假装裁剪到位） assert archived["archive_meta"]["result_bytes"] > budget def test_archive_meta_complete_when_under_budget(self, api) -> None: client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp = _exp_by_id(Session, resp.headers["X-Experiment-Id"]) assert exp is not None archived = json.loads(exp.result_json) meta = archived["archive_meta"] assert meta["truncated"] is False assert meta["curves_stored"] == meta["curves_total"] == len(archived["symbol_curves"]) assert meta["result_bytes"] == len(exp.result_json.encode("utf-8")) assert meta["result_chars"] == len(exp.result_json) class TestSingleResultCopy: """P1：完整结果只存 experiment 一份；GET /api/jobs/{id} 契约与老记录兼容。""" def test_submit_and_run_reads_through_in_memory_only(self, tmp_path) -> None: """`submit_and_run` 的返回对象仍带完整结果（既有调用方如 scripts/run_dividend_case.py 依赖它），但**数据库里不存第二份**。""" from app.application.services.job_executor import submit_and_run engine = create_engine(f"sqlite:///{tmp_path / 'submit.db'}", future=True) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine, expire_on_commit=False) drifts = {sym: 0.003 - 0.0015 * i for i, sym in enumerate(_SYMS)} bars = bars_dataframe_to_daily_bars(synthetic_daily(drifts, n=300)) stocks = [ Stock(symbol=sym, name=f"测试股份{i}", list_date=date(1999, 11, 10)) for i, sym in enumerate(_SYMS) ] spec = ResearchSpec.model_validate(_BACKTEST_BODY) done = submit_and_run( spec, factories={ "session_factory": Session, "job_repo_factory": lambda s: SqlAlchemyJobRepository(s), "experiment_repo_factory": lambda s: SqlAlchemyExperimentRepository(s), "stock_repo_factory": lambda s: _MemStockRepo(stocks), "daily_repo_factory": lambda s: _MemDailyRepo(bars), "engine": LocalEngine(), }, ) assert done.status == JobStatus.SUCCESS assert done.experiment_id assert done.result_json, "返回对象必须带完整结果（内存读透，供既有调用方使用）" assert BacktestResult.model_validate_json(done.result_json).symbol_curves with Session() as session: persisted = SqlAlchemyJobRepository(session).get(done.id) exp = SqlAlchemyExperimentRepository(session).get(done.experiment_id) assert persisted is not None and persisted.result_json is None, "数据库不得存第二份" assert exp is not None and exp.result_json, "完整结果在 experiment 侧" def test_job_view_reads_experiment_and_keeps_full_result(self, api) -> None: from app.api.jobs import _job_view client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp_id = resp.headers["X-Experiment-Id"] with Session() as session: job = JobRecord( id="JOB-VIEW-1", kind="backtest", spec_json=json.dumps(_BACKTEST_BODY), status=JobStatus.SUCCESS, result_json=None, # 新形态：job 侧不再保存结果副本 experiment_id=exp_id, created_at=datetime.now(), ) SqlAlchemyJobRepository(session).create(job) session.commit() view = _job_view(job, SqlAlchemyExperimentRepository(session)) assert view["result_source"] == "experiment" assert view["result"].symbol_curves, "契约不变：GET /api/jobs/{id} 仍返回完整结果" assert ( view["result"].summary.total_return_pct == resp.json()["summary"]["total_return_pct"] ) assert client.get("/api/jobs/JOB-VIEW-1").json()["result"]["symbol_curves"] def test_legacy_job_record_still_decodes_from_result_json(self, api) -> None: """老形态（result_json 有值、experiment_id 为空）必须继续可用。""" from app.api.jobs import _job_view client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) full = json.dumps(resp.json(), ensure_ascii=False) with Session() as session: job = JobRecord( id="JOB-LEGACY-1", kind="backtest", spec_json=json.dumps(_BACKTEST_BODY), status=JobStatus.SUCCESS, result_json=full, experiment_id=None, created_at=datetime.now(), ) SqlAlchemyJobRepository(session).create(job) session.commit() view = _job_view(job, SqlAlchemyExperimentRepository(session)) assert view["result_source"] == "job" assert view["result"].symbol_curves assert view["result"].summary.total_return_pct == resp.json()["summary"]["total_return_pct"] def test_job_table_no_longer_duplicates_result(self, api) -> None: """json 列确实没有第二份结果（体积审计的机器可读证据）。""" client, Session = api resp = client.post("/api/backtests", json=_BACKTEST_BODY) exp_id = resp.headers["X-Experiment-Id"] with Session() as session: job = JobRecord( id="JOB-SIZE-1", kind="backtest", spec_json=json.dumps(_BACKTEST_BODY), status=JobStatus.SUCCESS, experiment_id=exp_id, created_at=datetime.now(), ) SqlAlchemyJobRepository(session).create(job) session.commit() with Session() as session: row = session.execute( select(JobModel.result_json).where(JobModel.id == "JOB-SIZE-1") ).scalar() exp_row = session.execute( select(ExperimentModel.result_json).where(ExperimentModel.id == exp_id) ).scalar() assert row is None, "job 侧不得再存结果副本" assert exp_row, "完整结果必须在 experiment 侧" archived = _exp_by_id(Session, exp_id) assert archived is not None assert len(exp_row) == len(archived.result_json)