545 lines
19 KiB
Python
545 lines
19 KiB
Python
import hashlib
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import quant60.snapshot_backtest as snapshot_backtest_module
|
|
from quant60.config import BacktestConfig, StrategyConfig
|
|
from quant60.backtest import verify_artifact_manifest
|
|
from quant60.data_snapshot import (
|
|
CanonicalBarRecord,
|
|
IndexMembershipRecord,
|
|
build_snapshot_payload,
|
|
write_data_snapshot,
|
|
)
|
|
from quant60.snapshot_pipeline import (
|
|
SnapshotDecisionError,
|
|
build_snapshot_decision,
|
|
first_executable_window,
|
|
verify_snapshot_decision,
|
|
write_snapshot_decision,
|
|
)
|
|
from quant60.snapshot_backtest import run_snapshot_backtest
|
|
|
|
|
|
def business_days(start: date, count: int) -> list[date]:
|
|
values = []
|
|
current = start
|
|
while len(values) < count:
|
|
if current.weekday() < 5:
|
|
values.append(current)
|
|
current += timedelta(days=1)
|
|
return values
|
|
|
|
|
|
def build_provider_snapshot(
|
|
root: Path,
|
|
*,
|
|
st_symbol_on_last_session: str | None = None,
|
|
) -> tuple[str, date]:
|
|
frozen_calendar = business_days(date(2024, 1, 2), 31)
|
|
sessions = frozen_calendar[:-1]
|
|
next_trading_session = frozen_calendar[-1]
|
|
bars = []
|
|
memberships = []
|
|
symbols = ("600000.XSHG", "000001.XSHE")
|
|
for index, session in enumerate(sessions):
|
|
for symbol in symbols:
|
|
if symbol == "600000.XSHG":
|
|
raw = 10.0 * (1.005**index)
|
|
factor = 1.0
|
|
else:
|
|
unsplit = 20.0 * (1.01**index)
|
|
raw = unsplit if index < 15 else unsplit / 2.0
|
|
factor = 1.0 if index < 15 else 2.0
|
|
bars.append(
|
|
CanonicalBarRecord(
|
|
trading_date=session,
|
|
symbol=symbol,
|
|
open=raw * 0.999,
|
|
high=raw * 1.01,
|
|
low=raw * 0.99,
|
|
close=raw,
|
|
volume=1_000_000,
|
|
money=raw * 1_000_000,
|
|
paused=False,
|
|
source="fixture-provider",
|
|
is_st=(
|
|
symbol == st_symbol_on_last_session
|
|
and session == sessions[-1]
|
|
),
|
|
adjustment="none",
|
|
adjustment_factor=factor,
|
|
previous_close=raw / 1.005,
|
|
limit_up=raw * 1.10,
|
|
limit_down=raw * 0.90,
|
|
)
|
|
)
|
|
memberships.append(
|
|
IndexMembershipRecord(
|
|
effective_date=session,
|
|
index_symbol="000905.XSHG",
|
|
member_symbol=symbol,
|
|
source="fixture-provider",
|
|
)
|
|
)
|
|
payload = build_snapshot_payload(
|
|
bars=bars,
|
|
memberships=memberships,
|
|
provider="fixture-provider",
|
|
provider_version="1",
|
|
retrieved_at=datetime(2024, 3, 1, tzinfo=timezone.utc),
|
|
next_trading_session=next_trading_session,
|
|
query={
|
|
"index_symbol": "000905.XSHG",
|
|
"start_date": sessions[0].isoformat(),
|
|
"end_date": sessions[-1].isoformat(),
|
|
"trading_sessions": [
|
|
item.isoformat() for item in frozen_calendar
|
|
],
|
|
"adjustment": "none",
|
|
},
|
|
)
|
|
paths = write_data_snapshot(payload, root)
|
|
return paths["manifest"], sessions[-1]
|
|
|
|
|
|
class SnapshotPipelineTests(unittest.TestCase):
|
|
@staticmethod
|
|
def _broker_snapshot(
|
|
as_of: date,
|
|
*,
|
|
observed_at: datetime | None = None,
|
|
next_session: date = date(2024, 2, 13),
|
|
) -> dict:
|
|
signal_timestamp = f"{as_of.isoformat()}T15:00:00+08:00"
|
|
executable_window = first_executable_window(next_session)
|
|
observation_timestamp = (
|
|
observed_at.isoformat()
|
|
if observed_at is not None
|
|
else executable_window["start"]
|
|
)
|
|
return {
|
|
"schema_version": "1.0",
|
|
"snapshot_id": "broker-fixture",
|
|
"signal_as_of": signal_timestamp,
|
|
"next_trading_session": next_session.isoformat(),
|
|
"first_executable_window": executable_window,
|
|
"observation_as_of": observation_timestamp,
|
|
"as_of": observation_timestamp,
|
|
"source_time": observation_timestamp,
|
|
"broker": "fixture",
|
|
"account_hash": "fixture-account-hash",
|
|
"cash": 1_000_000,
|
|
"total_asset": 1_000_000,
|
|
"market_value": 0,
|
|
"positions": [],
|
|
"open_orders": [],
|
|
"metadata": {},
|
|
}
|
|
|
|
def test_next_day_broker_observation_binds_to_prior_close_signal(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(lookback=5, top_n=1),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
snapshot_manifest, as_of = build_provider_snapshot(
|
|
root / "snapshot"
|
|
)
|
|
observed_at = datetime(
|
|
as_of.year,
|
|
as_of.month,
|
|
as_of.day,
|
|
9,
|
|
0,
|
|
tzinfo=timezone(timedelta(hours=8)),
|
|
) + timedelta(days=1)
|
|
broker = self._broker_snapshot(
|
|
as_of,
|
|
observed_at=observed_at,
|
|
)
|
|
result = build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
broker_snapshot=broker,
|
|
)
|
|
self.assertEqual(
|
|
result["decision"]["signal_as_of"],
|
|
f"{as_of.isoformat()}T15:00:00+08:00",
|
|
)
|
|
self.assertEqual(
|
|
result["decision"]["broker_snapshot"][
|
|
"observation_as_of"
|
|
],
|
|
observed_at.isoformat(),
|
|
)
|
|
paths = write_snapshot_decision(
|
|
result,
|
|
root / "decision",
|
|
)
|
|
verified = verify_snapshot_decision(paths["manifest"])
|
|
self.assertEqual(
|
|
verified["signal_as_of"],
|
|
f"{as_of.isoformat()}T15:00:00+08:00",
|
|
)
|
|
|
|
def test_verified_snapshot_builds_deterministic_portable_target(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(
|
|
lookback=5,
|
|
top_n=1,
|
|
max_weight=0.50,
|
|
gross_target=0.50,
|
|
cash_buffer=0.10,
|
|
),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
snapshot_manifest, as_of = build_provider_snapshot(
|
|
root / "snapshot"
|
|
)
|
|
first = build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
equity=1_000_000,
|
|
)
|
|
second = build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
equity=1_000_000,
|
|
)
|
|
self.assertEqual(first, second)
|
|
self.assertEqual(len(first["signals"]), 2)
|
|
self.assertGreater(
|
|
next(
|
|
row["score"]
|
|
for row in first["signals"]
|
|
if row["symbol"] == "000001.XSHE"
|
|
),
|
|
next(
|
|
row["score"]
|
|
for row in first["signals"]
|
|
if row["symbol"] == "600000.XSHG"
|
|
),
|
|
)
|
|
selected = [
|
|
row
|
|
for row in first["targets"]
|
|
if row["target_weight"] > 0
|
|
]
|
|
self.assertEqual(
|
|
[row["symbol"] for row in selected],
|
|
["000001.XSHE"],
|
|
)
|
|
paths = write_snapshot_decision(first, root / "decision")
|
|
verified = verify_snapshot_decision(paths["manifest"])
|
|
self.assertTrue(verified["ok"])
|
|
self.assertEqual(verified["signal_count"], 2)
|
|
self.assertEqual(verified["target_count"], 2)
|
|
|
|
def test_pit_st_member_is_excluded_before_portable_ranking(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(
|
|
lookback=5,
|
|
top_n=1,
|
|
max_weight=0.50,
|
|
gross_target=0.50,
|
|
cash_buffer=0.10,
|
|
),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
snapshot_manifest, as_of = build_provider_snapshot(
|
|
Path(directory) / "snapshot",
|
|
st_symbol_on_last_session="000001.XSHE",
|
|
)
|
|
result = build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
equity=1_000_000,
|
|
)
|
|
self.assertEqual(
|
|
result["decision"]["universe"]["000001.XSHE"]["state"],
|
|
"excluded",
|
|
)
|
|
self.assertIn(
|
|
"ST_EXCLUDED",
|
|
result["decision"]["universe"]["000001.XSHE"]["reasons"],
|
|
)
|
|
self.assertNotIn(
|
|
"000001.XSHE",
|
|
result["decision"]["targets"],
|
|
)
|
|
|
|
def test_missing_exact_membership_fails_closed(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(lookback=5, top_n=1),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
snapshot_manifest, as_of = build_provider_snapshot(
|
|
Path(directory) / "snapshot"
|
|
)
|
|
with self.assertRaisesRegex(
|
|
SnapshotDecisionError,
|
|
"no next session",
|
|
):
|
|
build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of + timedelta(days=1),
|
|
equity=1_000_000,
|
|
)
|
|
|
|
def test_provider_snapshot_runs_local_event_backtest(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(
|
|
lookback=5,
|
|
top_n=1,
|
|
max_weight=0.50,
|
|
gross_target=0.50,
|
|
cash_buffer=0.10,
|
|
),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
snapshot_manifest, unused_as_of = build_provider_snapshot(
|
|
root / "snapshot"
|
|
)
|
|
del unused_as_of
|
|
with patch.object(
|
|
snapshot_backtest_module,
|
|
"load_verified_data_snapshot",
|
|
wraps=(
|
|
snapshot_backtest_module.load_verified_data_snapshot
|
|
),
|
|
) as snapshot_loader:
|
|
result = run_snapshot_backtest(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
)
|
|
self.assertEqual(snapshot_loader.call_count, 1)
|
|
self.assertGreater(result.report["decision_count"], 0)
|
|
self.assertEqual(
|
|
result.report["evidence_class"],
|
|
"provider-snapshot-local-backtest",
|
|
)
|
|
self.assertFalse(result.report["real_platform_backtest"])
|
|
self.assertFalse(result.report["investment_value_claim"])
|
|
self.assertGreater(len(result.signals), 0)
|
|
paths = result.write_artifacts(root / "backtest")
|
|
self.assertTrue(
|
|
verify_artifact_manifest(paths["manifest"])["ok"]
|
|
)
|
|
|
|
def test_decision_tampering_is_rejected(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(lookback=5, top_n=1),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
snapshot_manifest, as_of = build_provider_snapshot(
|
|
root / "snapshot"
|
|
)
|
|
result = build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
equity=1_000_000,
|
|
)
|
|
paths = write_snapshot_decision(result, root / "decision")
|
|
decision_path = Path(paths["decision"])
|
|
decision = json.loads(decision_path.read_text(encoding="utf-8"))
|
|
decision["orders"] = {}
|
|
decision_path.write_text(json.dumps(decision), encoding="utf-8")
|
|
with self.assertRaisesRegex(
|
|
SnapshotDecisionError,
|
|
"artifact hash mismatch",
|
|
):
|
|
verify_snapshot_decision(paths["manifest"])
|
|
|
|
def test_copied_input_calendar_cannot_be_rebound_by_artifact_hash(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(lookback=5, top_n=1),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
snapshot_manifest, as_of = build_provider_snapshot(
|
|
root / "snapshot"
|
|
)
|
|
result = build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
equity=1_000_000,
|
|
)
|
|
paths = write_snapshot_decision(result, root / "decision")
|
|
input_path = Path(paths["input_data_manifest"])
|
|
copied = json.loads(input_path.read_text(encoding="utf-8"))
|
|
copied["query"]["trading_sessions"][-1] = "2024-02-14"
|
|
input_path.write_text(
|
|
json.dumps(
|
|
copied,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
manifest_path = Path(paths["manifest"])
|
|
manifest = json.loads(
|
|
manifest_path.read_text(encoding="utf-8")
|
|
)
|
|
manifest["artifact_sha256"][
|
|
"input_data_manifest.json"
|
|
] = hashlib.sha256(input_path.read_bytes()).hexdigest()
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
manifest,
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaisesRegex(
|
|
SnapshotDecisionError,
|
|
"calendar manifest hash binding mismatch",
|
|
):
|
|
verify_snapshot_decision(manifest_path)
|
|
|
|
def test_wrong_window_stale_source_or_duplicate_broker_facts_fail_closed(self):
|
|
config = BacktestConfig(
|
|
symbols=("600000.XSHG", "000001.XSHE"),
|
|
strategy=StrategyConfig(lookback=5, top_n=1),
|
|
)
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
snapshot_manifest, as_of = build_provider_snapshot(
|
|
Path(directory) / "snapshot"
|
|
)
|
|
observed_at = datetime(
|
|
as_of.year,
|
|
as_of.month,
|
|
as_of.day,
|
|
9,
|
|
2,
|
|
tzinfo=timezone(timedelta(hours=8)),
|
|
) + timedelta(days=1)
|
|
for source_time, message in (
|
|
(
|
|
(observed_at - timedelta(seconds=61)).isoformat(),
|
|
"more than 60s stale",
|
|
),
|
|
(
|
|
(observed_at + timedelta(seconds=6)).isoformat(),
|
|
"more than 5s after",
|
|
),
|
|
):
|
|
with self.subTest(source_time=source_time):
|
|
broker = self._broker_snapshot(
|
|
as_of,
|
|
observed_at=observed_at,
|
|
)
|
|
broker["source_time"] = source_time
|
|
with self.assertRaisesRegex(
|
|
SnapshotDecisionError,
|
|
message,
|
|
):
|
|
build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
broker_snapshot=broker,
|
|
)
|
|
signal_time = datetime(
|
|
as_of.year,
|
|
as_of.month,
|
|
as_of.day,
|
|
15,
|
|
0,
|
|
tzinfo=timezone(timedelta(hours=8)),
|
|
)
|
|
for observed_at, message in (
|
|
(
|
|
signal_time + timedelta(minutes=1),
|
|
"outside the unique next session",
|
|
),
|
|
(
|
|
observed_at.replace(hour=9, minute=30),
|
|
"outside the unique next session",
|
|
),
|
|
(
|
|
observed_at + timedelta(days=3),
|
|
"outside the unique next session",
|
|
),
|
|
):
|
|
with self.subTest(observed_at=observed_at):
|
|
broker = self._broker_snapshot(
|
|
as_of,
|
|
observed_at=observed_at,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
SnapshotDecisionError,
|
|
message,
|
|
):
|
|
build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
broker_snapshot=broker,
|
|
)
|
|
for field in ("signal_as_of", "observation_as_of"):
|
|
with self.subTest(missing=field):
|
|
broker = self._broker_snapshot(as_of)
|
|
del broker[field]
|
|
with self.assertRaisesRegex(
|
|
SnapshotDecisionError,
|
|
rf"missing required fields.*{field}",
|
|
):
|
|
build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
broker_snapshot=broker,
|
|
)
|
|
duplicate = self._broker_snapshot(as_of)
|
|
position = {
|
|
"symbol": "600000.XSHG",
|
|
"quantity": 100,
|
|
"sellable_quantity": 100,
|
|
"average_cost": 10,
|
|
"market_value": 1000,
|
|
}
|
|
duplicate["positions"] = [position, dict(position)]
|
|
with self.assertRaisesRegex(
|
|
SnapshotDecisionError,
|
|
"duplicate position",
|
|
):
|
|
build_snapshot_decision(
|
|
snapshot_manifest=snapshot_manifest,
|
|
config=config,
|
|
as_of=as_of,
|
|
broker_snapshot=duplicate,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|