"""Strategy reports preserve real ledger facts without factor-score fabrication.""" import json from dataclasses import asdict import numpy as np import pandas as pd import pytest from quant_engine.strategy_artifact import build_strategy_research_artifact from quant_engine.strategy_optimizer import optimize_strategy_research from quant_engine.strategy_research import BenchmarkInput, run_strategy_research def bars(closes, opens=None): close = np.asarray(closes, dtype=float) opening = np.asarray(opens if opens is not None else closes, dtype=float) return pd.DataFrame( { "open": opening, "high": np.maximum(close, opening), "low": np.minimum(close, opening), "close": close, }, index=pd.date_range("2026-01-01", periods=len(close), freq="B"), ) def run(strategy="BuyAndHold", feed=None, **kwargs): return run_strategy_research( strategy, bars([10, 11, 12, 13]) if feed is None else feed, asset="SYNTHETIC", initial_cash=1000, **kwargs, ) def build(result, **kwargs): metadata = { "run_id": "strategy-run", "strategy_id": "isolated-strategy", "strategy_name": "Synthetic", "strategy_version": "1", "engine_version": "candidate", "code_revision": "candidate", "data_snapshot_id": "synthetic:ohlc-v1", "calendar": "synthetic-sessions", "timezone": "Asia/Shanghai", "started_at": "2026-01-08T10:00:00+08:00", "finished_at": "2026-01-08T10:00:01+08:00", "parameters": {"synthetic": True}, } return build_strategy_research_artifact(result, **(metadata | kwargs)) def report(artifact): return json.loads(artifact.run.iloc[0]["params_json"])["strategy_report"] def test_projects_same_ledger_cash_fees_positions_and_completed_trade_basis(): result = run( "SmaCross", bars([10, 8, 12, 6, 14, 5, 12]), params={"fast": 1, "slow": 2}, commission=0.01, stamp_duty=0.02, ) artifact = build(result) detail = report(artifact) assert artifact.schema_version == "1.1.0" assert artifact.nav.portfolio_value.tolist() == result.ledger.nav_series.tolist() assert artifact.nav.pnl_pct.tolist() == result.ledger.daily_returns.tolist() assert artifact.trades.fee.sum() == pytest.approx(result.ledger.trades_frame.fee.sum()) assert artifact.nav.total_cost.sum() == pytest.approx(artifact.trades.total_cost.sum()) assert artifact.performance.iloc[0].win_rate == result.pairing.win_rate assert artifact.performance.iloc[0].n_trades == len(result.ledger.trades_frame) assert detail["trade_pairing"] == json.loads(json.dumps(asdict(result.pairing))) assert detail["costs"] == { "initial_cash": 1000, "commission": 0.01, "stamp_duty": 0.02, "min_trade_amount": 0, "slippage_bps": 0, } assert detail["decision_eligible"] is False for day, frame in artifact.positions.groupby("trade_date"): nav = artifact.nav.loc[artifact.nav.trade_date == day].iloc[0] assert frame.market_value.sum() == pytest.approx(nav.portfolio_value) assert frame.weight.sum() == pytest.approx(1) assert artifact.signals.empty assert artifact.attribution.empty assert artifact.risk.empty assert detail["projections"]["signals"] == "strategy_report.signals" assert detail["projections"]["attribution"] == "not_computed" params = json.loads(artifact.run.iloc[0].params_json) assert params["performance_interpretation"]["win_rate_basis"] == "completed_trades" def test_last_signal_is_not_lost_or_fabricated_as_a_factor_signal(): result = run("SmaCross", bars([10, 8, 12]), params={"fast": 1, "slow": 2}) artifact = build(result) signal = report(artifact)["signals"][0] assert signal["status"] == "no_next_session" assert signal["execution_date"] is None assert signal["signal_id"] == "strategy-run:signal:2026-01-05" assert artifact.trades.empty assert artifact.signals.empty def test_signal_ids_join_actual_fills_and_report(): artifact = build(run()) signals = {item["signal_id"]: item for item in report(artifact)["signals"]} for fill in artifact.trades.to_dict("records"): signal = signals[fill["signal_id"]] assert signal["execution_date"] == fill["trade_date"].isoformat() assert signal["decision_date"] < signal["execution_date"] @pytest.mark.parametrize( "name", [ "BuyAndHold", "SmaCross", "MACross", "RSI", "BollingerBreakout", "DualThrust", "TurtleBreakout", ], ) def test_all_seven_defaults_have_canonical_serializable_reports(name): values = 10 + np.sin(np.arange(80) / 2) * 2 result = run(name, bars(values, np.r_[values[0], values[:-1]])) artifact = build(result) assert report(artifact)["parameters"] == result.parameters assert json.loads(artifact.canonical_json())["schema_version"] == "1.1.0" assert artifact.content_sha256 == build(result).content_sha256 assert len(artifact.nav) == 80 @pytest.mark.parametrize("status", ["not_requested", "empty", "present"]) def test_benchmark_states_and_original_returns_are_preserved(status): feed = bars([10, 11, 12, 13]) closes = ( pd.Series([20, 22, 21, 23], index=feed.index) if status == "present" else (pd.Series(dtype=float) if status == "empty" else None) ) result = run(feed=feed, benchmark=BenchmarkInput(status, closes)) artifact = build(result, benchmark_id="SYNTHETIC-BENCH" if status != "not_requested" else None) assert report(artifact)["benchmark"]["status"] == status if status == "present": assert artifact.nav.benchmark_return.tolist() == pytest.approx( [0, 0.1, 21 / 22 - 1, 23 / 21 - 1] ) assert artifact.nav.benchmark_nav.tolist() == pytest.approx([1, 1.1, 1.05, 1.15]) assert artifact.run.iloc[0].benchmark_alignment_policy == "exact_session_index" else: assert artifact.nav.benchmark_nav.isna().all() assert artifact.run.iloc[0].benchmark_alignment_policy == "none" def test_zero_nav_preserves_zero_value_and_undefined_weight_with_reason(): result = run( "SmaCross", bars([10, 8, 12, 6, 14]), params={"fast": 1, "slow": 2}, commission=0, stamp_duty=1, ) artifact = build(result) last = artifact.positions.iloc[-1] assert last.market_value == 0 assert pd.isna(last.weight) assert artifact.nav.iloc[-1].nav == 0 assert report(artifact)["projections"]["undefined_weight_dates"] == ["2026-01-07"] def test_no_closed_lot_metrics_are_null_with_covered_reason(): artifact = build(run()) metadata = json.loads(artifact.run.iloc[0].params_json) assert report(artifact)["metrics"]["trade_win_rate"] is None assert ( metadata["performance_interpretation"]["unavailable_reasons"]["win_rate"] == "no_closed_lots" ) assert pd.isna(artifact.performance.iloc[0].win_rate) def test_result_snapshots_detach_caller_data_and_returned_views(): feed = bars([10, 11, 12, 13]) result = run(feed=feed) before = build(result).content_sha256 feed.iloc[:] = 999 view = result.bars view.iloc[:] = 777 artifact = build(result) assert artifact.content_sha256 == before positions = artifact.positions positions["market_value"] = 0 assert artifact.content_sha256 == before def test_grid_artifact_retains_all_ranks_and_selected_ledger_and_detaches_input(): grid = {"buy_pct": [0.2, 0.5, 1]} result = optimize_strategy_research( "BuyAndHold", bars([10, 11, 12, 13]), asset="SYNTHETIC", param_grid=grid, objective="total_return", initial_cash=1000, commission=0, stamp_duty=0, ) grid["buy_pct"].append(0.9) artifact = build(result) ranking = report(artifact)["optimization"] assert ranking["grid"] == {"buy_pct": [0.2, 0.5, 1]} assert ranking["trial_count"] == 3 assert ranking["selected_rank"] == 1 assert [trial["rank"] for trial in ranking["trials"]] == [1, 2, 3] assert [trial["score"] for trial in ranking["trials"]] == [ trial.score for trial in result.trials ] assert ranking["trials"][0]["parameters"] == {"buy_pct": 1} assert ( artifact.nav.portfolio_value.tolist() == result.trials[0].result.ledger.nav_series.tolist() ) assert all("ledger" not in trial for trial in ranking["trials"]) @pytest.mark.parametrize("key", ["strategy_report", "performance_interpretation"]) def test_callers_cannot_overwrite_authoritative_report_or_metric_explanation(key): with pytest.raises(ValueError, match="reserved"): build(run(), parameters={key: {"decision_eligible": True}}) def test_report_mutation_changes_canonical_artifact_digest(): first = build(run(commission=0)) second = build(run(commission=0.01)) assert first.content_sha256 != second.content_sha256 assert first.run.iloc[0].config_hash != second.run.iloc[0].config_hash def test_invalid_metadata_fails_before_artifact_creation(): with pytest.raises(ValueError, match="finished_at"): build(run(), finished_at="2026-01-07T10:00:00+08:00") with pytest.raises(ValueError, match="benchmark"): build(run(), benchmark_id="FAKE") def test_missing_relative_metrics_explain_their_fact_column_names(): artifact = build(run()) reasons = json.loads(artifact.run.iloc[0].params_json)["performance_interpretation"][ "unavailable_reasons" ] assert reasons["ir"] == "benchmark_not_requested" assert "information_ratio" not in reasons @pytest.mark.parametrize("producer", ["run", "optimization", "artifact"]) def test_reserved_cash_asset_cannot_collide_with_cash_position(producer): from dataclasses import replace operation = { "run": lambda: run_strategy_research("BuyAndHold", bars([10, 11]), asset="CASH"), "optimization": lambda: optimize_strategy_research( "BuyAndHold", bars([10, 11]), asset="CASH", param_grid={"buy_pct": [0.5]}, objective="total_return", ), "artifact": lambda: build(replace(run(params={"buy_pct": 0}), asset="CASH")), }[producer] with pytest.raises(ValueError, match="asset"): operation() def test_full_100_trial_tied_grid_keeps_complete_stable_ranking(): grid = {"k1": [index / 10 for index in range(10)], "k2": [index / 10 for index in range(10)]} result = optimize_strategy_research( "DualThrust", bars([10] * 10), asset="SYNTHETIC", param_grid=grid, objective="total_return" ) ranking = report(build(result))["optimization"] assert ranking["trial_count"] == 100 assert len(ranking["trials"]) == 100 assert [trial["parameters"] for trial in ranking["trials"]] == [ trial.parameters for trial in result.trials ] assert all(trial["score"] == 0 for trial in ranking["trials"])