diff --git a/infra/control_center/.env.example b/infra/control_center/.env.example new file mode 100644 index 0000000..a8dc6fc --- /dev/null +++ b/infra/control_center/.env.example @@ -0,0 +1 @@ +POSTGRES_PASSWORD=replace_me diff --git a/infra/control_center/README.md b/infra/control_center/README.md new file mode 100644 index 0000000..2849598 --- /dev/null +++ b/infra/control_center/README.md @@ -0,0 +1,3 @@ +# QuantPilot Control Center + +Start with `cp .env.example .env && docker compose up -d`; stop with `docker compose down` (add `-v` to remove local reporting data). Grafana is read-only and uses PostgreSQL as its sole datasource. diff --git a/infra/control_center/docker-compose.yml b/infra/control_center/docker-compose.yml new file mode 100644 index 0000000..5f9f464 --- /dev/null +++ b/infra/control_center/docker-compose.yml @@ -0,0 +1,24 @@ +services: + postgres: + image: postgres:16 + environment: + POSTGRES_DB: quantpilot + POSTGRES_USER: quantpilot + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?set in .env} + ports: ["5432:5432"] + volumes: ["pgdata:/var/lib/postgresql/data"] + healthcheck: + test: ["CMD-SHELL", "pg_isready -U quantpilot -d quantpilot"] + interval: 10s + timeout: 5s + retries: 5 + grafana: + image: grafana/grafana:11.2.0 + ports: ["3000:3000"] + volumes: + - ./grafana/provisioning:/etc/grafana/provisioning:ro + - ./grafana/dashboards:/var/lib/grafana/dashboards:ro + depends_on: + postgres: + condition: service_healthy +volumes: {pgdata: {}} diff --git a/infra/control_center/grafana/dashboards/ai-firm-learning-desk.json b/infra/control_center/grafana/dashboards/ai-firm-learning-desk.json new file mode 100644 index 0000000..af85b8c --- /dev/null +++ b/infra/control_center/grafana/dashboards/ai-firm-learning-desk.json @@ -0,0 +1 @@ +{"title":"AI Firm and Learning Desk","schemaVersion":39,"panels":[{"title":"Shadow calls cache and cost","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, mode, physical_model_calls, cache_hits, estimated_api_cost FROM ai_shadow_reports ORDER BY session_id DESC","format":"table"}]},{"title":"Learning Desk reports","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, payload FROM learning_desk_reports ORDER BY session_id DESC","format":"table"}]},{"title":"Shadow abstentions","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, payload->>'abstentions' AS abstentions FROM ai_shadow_reports ORDER BY session_id DESC","format":"table"}]}]} diff --git a/infra/control_center/grafana/dashboards/operations.json b/infra/control_center/grafana/dashboards/operations.json new file mode 100644 index 0000000..e54d810 --- /dev/null +++ b/infra/control_center/grafana/dashboards/operations.json @@ -0,0 +1 @@ +{"title":"Operations","schemaVersion":39,"panels":[{"title":"Session and Prefect IDs","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, run_id, prefect_flow_id, prefect_deployment_id, prefect_flow_run_id, status FROM paper_sessions ORDER BY completed_at DESC","format":"table"}]},{"title":"Replay and state hashes","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, replay_status, state_hash_before, state_hash_after FROM paper_sessions ORDER BY completed_at DESC","format":"table"}]},{"title":"Failure classification","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, failure_class, failure_reason FROM paper_sessions WHERE failure_class IS NOT NULL ORDER BY completed_at DESC","format":"table"}]}]} diff --git a/infra/control_center/grafana/dashboards/orders-and-execution.json b/infra/control_center/grafana/dashboards/orders-and-execution.json new file mode 100644 index 0000000..3b16930 --- /dev/null +++ b/infra/control_center/grafana/dashboards/orders-and-execution.json @@ -0,0 +1 @@ +{"title":"Orders and Execution","schemaVersion":39,"panels":[{"title":"Order intents","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, source_identity, payload FROM paper_orders ORDER BY session_id DESC","format":"table"}]},{"title":"Fills","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, source_identity, payload FROM paper_fills ORDER BY session_id DESC","format":"table"}]},{"title":"Order and fill counts","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT s.session_id, count(DISTINCT o.source_identity) AS order_count, count(DISTINCT f.source_identity) AS fill_count FROM paper_sessions s LEFT JOIN paper_orders o ON o.session_id=s.session_id LEFT JOIN paper_fills f ON f.session_id=s.session_id GROUP BY s.session_id","format":"table"}]}]} diff --git a/infra/control_center/grafana/dashboards/portfolio-and-pnl.json b/infra/control_center/grafana/dashboards/portfolio-and-pnl.json new file mode 100644 index 0000000..e3c9cbc --- /dev/null +++ b/infra/control_center/grafana/dashboards/portfolio-and-pnl.json @@ -0,0 +1 @@ +{"title":"Portfolio and PnL","schemaVersion":39,"panels":[{"title":"Source equity curve","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, payload FROM paper_equity_curve ORDER BY session_id DESC","format":"table"}]},{"title":"Source positions","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, source_identity AS symbol, payload FROM paper_positions ORDER BY session_id DESC","format":"table"}]},{"title":"Reconciliation source facts","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT session_id, payload FROM paper_reconciliation ORDER BY session_id DESC","format":"table"}]}]} diff --git a/infra/control_center/grafana/dashboards/quantpilot-overview.json b/infra/control_center/grafana/dashboards/quantpilot-overview.json new file mode 100644 index 0000000..ace4d4f --- /dev/null +++ b/infra/control_center/grafana/dashboards/quantpilot-overview.json @@ -0,0 +1 @@ +{"title":"QuantPilot Overview","schemaVersion":39,"panels":[{"title":"Latest status and manifest","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT status, production_manifest_digest, production_input_digest FROM paper_sessions ORDER BY completed_at DESC LIMIT 1","format":"table"}]},{"title":"Latest equity facts","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT payload FROM paper_equity_curve ORDER BY session_id DESC LIMIT 20","format":"table"}]},{"title":"Positions and reconciliation","type":"table","datasource":{"uid":"quantpilot-postgres"},"targets":[{"rawSql":"SELECT p.payload AS position, r.payload AS reconciliation FROM paper_positions p LEFT JOIN paper_reconciliation r ON r.session_id=p.session_id ORDER BY p.session_id DESC","format":"table"}]}]} diff --git a/infra/control_center/grafana/provisioning/dashboards/quantpilot.yml b/infra/control_center/grafana/provisioning/dashboards/quantpilot.yml new file mode 100644 index 0000000..5df231d --- /dev/null +++ b/infra/control_center/grafana/provisioning/dashboards/quantpilot.yml @@ -0,0 +1,6 @@ +apiVersion: 1 +providers: + - name: QuantPilot + folder: QuantPilot + type: file + options: {path: /var/lib/grafana/dashboards} diff --git a/infra/control_center/grafana/provisioning/datasources/quantpilot.yml b/infra/control_center/grafana/provisioning/datasources/quantpilot.yml new file mode 100644 index 0000000..507ba3a --- /dev/null +++ b/infra/control_center/grafana/provisioning/datasources/quantpilot.yml @@ -0,0 +1,11 @@ +apiVersion: 1 +datasources: + - name: QuantPilot PostgreSQL + uid: quantpilot-postgres + type: postgres + access: proxy + isDefault: true + url: postgres:5432 + user: quantpilot + secureJsonData: {password: "${POSTGRES_PASSWORD}"} + jsonData: {database: quantpilot, sslmode: disable, postgresVersion: 1600} diff --git a/pyproject.toml b/pyproject.toml index 350bf93..947a0ad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,10 @@ replay = [ ml = [ "lightgbm>=4,<5", ] +continuous-paper = [ + "psycopg[binary]>=3.1,<4", + "prefect>=3,<4", +] [tool.setuptools.packages.find] where = ["src"] diff --git a/scripts/run_continuous_paper_cycle_v1.py b/scripts/run_continuous_paper_cycle_v1.py new file mode 100644 index 0000000..1fac0bc --- /dev/null +++ b/scripts/run_continuous_paper_cycle_v1.py @@ -0,0 +1,33 @@ +#!/usr/bin/env python3 +"""Manual, offline-safe entry point for one continuous paper cycle.""" +from __future__ import annotations +import argparse +import os +from pathlib import Path +from quantpilot_core.continuous_paper import ActiveShadowConfig, ContinuousPaperCycle, ContinuousPaperCycleConfig, InMemoryReportingStore, PostgreSQLReportingStore +from quantpilot_core.production_candidate import load_runtime_manifest +from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig + + +def main() -> None: + p = argparse.ArgumentParser() + p.add_argument("--production-manifest", required=True); p.add_argument("--decision-session", required=True) + p.add_argument("--state-path", required=True); p.add_argument("--report-path", required=True) + p.add_argument("--dsn-env-var", default="QUANTPILOT_POSTGRES_DSN"); p.add_argument("--test-store", action="store_true") + p.add_argument("--cache-path", default=".cache/quantpilot/active_shadow"); p.add_argument("--timeout", type=float, default=30.0) + p.add_argument("--allowed-roles", default=""); p.add_argument("--enable-active-shadow", action="store_true"); p.add_argument("--enable-live-ai", action="store_true") + p.add_argument("--max-physical-model-calls", type=int, default=0); p.add_argument("--max-estimated-api-cost", type=float, default=0.0); p.add_argument("--estimated-cost-per-call", type=float, default=0.0) + args = p.parse_args() + if args.enable_live_ai and (not args.enable_active_shadow or args.max_physical_model_calls <= 0 or args.max_estimated_api_cost <= 0 or args.estimated_cost_per_call <= 0): + p.error("live AI requires active shadow and positive explicit call, cycle-cost, and per-call cost limits") + dsn = os.environ.get(args.dsn_env_var) + if not args.test_store and not dsn: p.error("set the DSN environment variable or use explicit --test-store") + manifest = load_runtime_manifest(Path(args.production_manifest)) + pipeline = RealCandidatePipelineConfig(decision_session=args.decision_session, state_path=args.state_path, report_path=args.report_path, production_manifest=manifest) + store = InMemoryReportingStore() if args.test_store else PostgreSQLReportingStore(dsn) + shadow = ActiveShadowConfig(enabled=args.enable_active_shadow, enable_live_calls=args.enable_live_ai, max_physical_model_calls_per_cycle=args.max_physical_model_calls, max_estimated_cost_per_cycle=args.max_estimated_api_cost, estimated_cost_per_call=args.estimated_cost_per_call, cache_path=args.cache_path, timeout=args.timeout) + result = ContinuousPaperCycle(store).run_once(ContinuousPaperCycleConfig(pipeline, run_id=f"manual-{args.decision_session}", active_shadow=shadow)) + print(f"session_id={result.session_id} status={result.status} run_id={result.run_id}") + + +if __name__ == "__main__": main() diff --git a/scripts/serve_continuous_paper_v1.py b/scripts/serve_continuous_paper_v1.py new file mode 100644 index 0000000..783f00a --- /dev/null +++ b/scripts/serve_continuous_paper_v1.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +"""Explicit Prefect deployment helper; no schedule starts during import.""" +from __future__ import annotations +import argparse + + +def parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser() + p.add_argument("--production-manifest", required=True) + p.add_argument("--state-path", required=True) + p.add_argument("--report-path", required=True) + p.add_argument("--dsn-env-var", default="QUANTPILOT_POSTGRES_DSN") + p.add_argument("--cron", default="30 16 * * 1-5") + p.add_argument("--timezone", default="Asia/Shanghai") + p.add_argument("--max-physical-model-calls", type=int, default=0) + p.add_argument("--max-estimated-api-cost", type=float, default=0.0) + p.add_argument("--estimated-cost-per-call", type=float, default=0.0) + return p + + +def main() -> None: + args = parser().parse_args() + # Deliberately lazy: --help stays useful in minimal/offline environments. + from quantpilot_core.continuous_paper.prefect_flow import continuous_paper_flow + if not hasattr(continuous_paper_flow, "serve"): + raise RuntimeError("install quantpilot-ai-next[continuous-paper] to serve a Prefect deployment") + parameters = {"production_manifest_path": args.production_manifest, "state_path": args.state_path, "report_path": args.report_path, "dsn_env_var": args.dsn_env_var, "active_shadow": {"max_physical_model_calls_per_cycle": args.max_physical_model_calls, "max_estimated_cost_per_cycle": args.max_estimated_api_cost, "estimated_cost_per_call": args.estimated_cost_per_call}} + continuous_paper_flow.serve(name="quantpilot-continuous-paper-v1", cron=args.cron, timezone=args.timezone, parameters=parameters) + + +if __name__ == "__main__": main() diff --git a/src/quantpilot_core/continuous_paper/__init__.py b/src/quantpilot_core/continuous_paper/__init__.py new file mode 100644 index 0000000..046ee9b --- /dev/null +++ b/src/quantpilot_core/continuous_paper/__init__.py @@ -0,0 +1,13 @@ +"""One-cycle, advisory-only continuous paper orchestration.""" + +from .active_shadow import ActiveShadowConfig, ActiveShadowRunner +from .contracts import ContinuousPaperCycleConfig, ContinuousPaperCycleResult, ReportingCycleBundle +from .service import ContinuousPaperCycle +from .store import InMemoryReportingStore, PostgreSQLReportingStore, ReportingConflictError + +__all__ = [ + "ActiveShadowConfig", "ActiveShadowRunner", "ContinuousPaperCycle", + "ContinuousPaperCycleConfig", "ContinuousPaperCycleResult", "InMemoryReportingStore", + "PostgreSQLReportingStore", "ReportingConflictError", + "ReportingCycleBundle", +] diff --git a/src/quantpilot_core/continuous_paper/active_shadow.py b/src/quantpilot_core/continuous_paper/active_shadow.py new file mode 100644 index 0000000..94b311d --- /dev/null +++ b/src/quantpilot_core/continuous_paper/active_shadow.py @@ -0,0 +1,123 @@ +"""Bounded, advisory-only active shadow. It cannot affect production execution.""" +from __future__ import annotations + +import json +import os +from dataclasses import asdict, dataclass, is_dataclass +from pathlib import Path +from typing import Any, Mapping + +from quantpilot_core.quant_firm.deepseek_advisory import DeepSeekAdvisoryAgent, DeepSeekAdvisoryInput, DeepSeekAdvisoryRole, DeepSeekClientConfig +from .store import payload_digest + + +@dataclass(frozen=True) +class ActiveShadowConfig: + enabled: bool = False + enable_live_calls: bool = False + max_physical_model_calls_per_cycle: int = 0 + max_estimated_cost_per_cycle: float = 0.0 + estimated_cost_per_call: float = 0.0 + cache_path: str | Path = ".cache/quantpilot/active_shadow" + allow_roles: tuple[DeepSeekAdvisoryRole, ...] = tuple(DeepSeekAdvisoryRole) + timeout: float = 30.0 + failure_policy: str = "abstain" + + def __post_init__(self) -> None: + if self.failure_policy != "abstain": + raise ValueError("active shadow supports only failure_policy='abstain'") + if self.timeout <= 0: + raise ValueError("timeout must be positive") + if self.enable_live_calls and (self.max_physical_model_calls_per_cycle <= 0 or self.max_estimated_cost_per_cycle <= 0 or self.estimated_cost_per_call <= 0): + raise ValueError("live active shadow requires positive call, cycle-cost, and per-call cost limits") + + +class ActiveShadowRunner: + def __init__(self, config: ActiveShadowConfig, *, agent: Any | None = None) -> None: + self.config = config + self.agent = agent or DeepSeekAdvisoryAgent(DeepSeekClientConfig(timeout_seconds=config.timeout, enable_live_call=config.enable_live_calls)) + + def run(self, evidence: Mapping[str, Any]) -> Mapping[str, Any]: + evidence_digest = payload_digest(evidence) + result: dict[str, Any] = {"mode": "disabled", "evidence_digest": evidence_digest, "physical_model_calls": 0, "cache_hits": 0, "estimated_api_cost": 0.0, "cost_is_estimated": True, "input_tokens": 0, "output_tokens": 0, "abstentions": 0, "roles": {}} + if not self.config.enabled: + return result + result["mode"] = "active_shadow" + root = Path(self.config.cache_path) + root.mkdir(parents=True, exist_ok=True) + roles = tuple(role for role in self.config.allow_roles if role is not DeepSeekAdvisoryRole.INVESTMENT_COMMITTEE) + if DeepSeekAdvisoryRole.INVESTMENT_COMMITTEE in self.config.allow_roles: + roles += (DeepSeekAdvisoryRole.INVESTMENT_COMMITTEE,) + for role in roles: + key = payload_digest({"role": role.value, "pit_evidence_digest": evidence_digest, "model": self._model_name(), "policy": "quant_firm_approved_v1"}) + cached = self._read_cache(root / f"{key}.json", key) + if cached is not None: + result["roles"][role.value] = cached + result["cache_hits"] += 1 + continue + if not self.config.enable_live_calls: + self._abstain(result, role, "live_calls_disabled") + continue + if result["physical_model_calls"] >= self.config.max_physical_model_calls_per_cycle: + self._abstain(result, role, "call_budget_exhausted") + continue + if result["estimated_api_cost"] + self.config.estimated_cost_per_call > self.config.max_estimated_cost_per_cycle: + self._abstain(result, role, "cost_budget_exhausted") + continue + # This count occurs before the physical request, including provider failures. + result["physical_model_calls"] += 1 + result["estimated_api_cost"] += self.config.estimated_cost_per_call + try: + role_evidence: Mapping[str, Any] = evidence + if role is DeepSeekAdvisoryRole.INVESTMENT_COMMITTEE: + specialist_digests = tuple( + payload_digest(payload) + for specialist, payload in result["roles"].items() + if specialist != role.value and isinstance(payload, Mapping) and payload.get("status") != "abstain" + ) + role_evidence = {**evidence, "specialist_evidence_digests": specialist_digests} + output = self.agent.advise(DeepSeekAdvisoryInput(role=role, quant_firm_decision_report_summary=role_evidence, learning_desk_output=evidence.get("learning_desk"), run_label=str(evidence.get("session_id", "")))) + payload = _json(output) + if getattr(output, "is_fallback", False) or not isinstance(payload, Mapping): + raise ValueError("invalid advisory output") + result["input_tokens"] += int(payload.get("input_tokens", 0) or 0) + result["output_tokens"] += int(payload.get("output_tokens", 0) or 0) + self._write_cache(root / f"{key}.json", {"cache_key": key, "payload_digest": payload_digest(payload), "payload": payload}) + result["roles"][role.value] = payload + except Exception as exc: + self._abstain(result, role, f"provider_or_validation_failure:{type(exc).__name__}") + return result + + def _model_name(self) -> str: + return str(getattr(getattr(self.agent, "config", None), "model", "approved-default")) + + @staticmethod + def _abstain(result: dict[str, Any], role: DeepSeekAdvisoryRole, reason: str) -> None: + result["roles"][role.value] = {"status": "abstain", "reason": reason} + result["abstentions"] += 1 + + @staticmethod + def _read_cache(path: Path, expected_key: str) -> Mapping[str, Any] | None: + try: + item = json.loads(path.read_text(encoding="utf-8")) + payload = item["payload"] + if item.get("cache_key") != expected_key or item.get("payload_digest") != payload_digest(payload) or not isinstance(payload, Mapping): + return None + return payload + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError): + return None + + @staticmethod + def _write_cache(path: Path, value: Mapping[str, Any]) -> None: + temporary = path.with_suffix(".tmp") + with temporary.open("w", encoding="utf-8") as handle: + json.dump(value, handle, sort_keys=True) + handle.flush(); os.fsync(handle.fileno()) + os.replace(temporary, path) + + +def _json(value: Any) -> Any: + if is_dataclass(value) and not isinstance(value, type): return _json(asdict(value)) + if isinstance(value, Mapping): return {str(k): _json(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): return [_json(v) for v in value] + return getattr(value, "value", value) diff --git a/src/quantpilot_core/continuous_paper/contracts.py b/src/quantpilot_core/continuous_paper/contracts.py new file mode 100644 index 0000000..700844d --- /dev/null +++ b/src/quantpilot_core/continuous_paper/contracts.py @@ -0,0 +1,51 @@ +"""Public, immutable contracts for one continuous paper-trading cycle.""" +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Mapping + +from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig + + +@dataclass(frozen=True) +class ContinuousPaperCycleConfig: + """Inputs owned by orchestration, never by the production strategy.""" + + pipeline_config: RealCandidatePipelineConfig + run_id: str + prefect_flow_id: str | None = None + prefect_deployment_id: str | None = None + prefect_flow_run_id: str | None = None + active_shadow: Any | None = None + + def __post_init__(self) -> None: + if not isinstance(self.run_id, str) or not self.run_id.strip(): + raise ValueError("run_id must be a non-empty string") + if not isinstance(self.pipeline_config, RealCandidatePipelineConfig): + raise TypeError("pipeline_config must be RealCandidatePipelineConfig") + + +@dataclass(frozen=True) +class ContinuousPaperCycleResult: + session_id: str + run_id: str + status: str + report_digest: str + production_input_digest: str + started_at: datetime + completed_at: datetime + replay_status: str + shadow_report: Mapping[str, Any] = field(default_factory=dict) + + +@dataclass(frozen=True) +class ReportingCycleBundle: + """All immutable reporting rows for one production session.""" + + session: Mapping[str, Any] + report: Mapping[str, Any] + learning: Mapping[str, Any] + shadow: Mapping[str, Any] + facts: Mapping[str, tuple[Mapping[str, Any], ...]] + diff --git a/src/quantpilot_core/continuous_paper/facts.py b/src/quantpilot_core/continuous_paper/facts.py new file mode 100644 index 0000000..7dab7b2 --- /dev/null +++ b/src/quantpilot_core/continuous_paper/facts.py @@ -0,0 +1,47 @@ +"""Extract source facts from a daily-paper report without recalculation.""" +from __future__ import annotations + +from typing import Any, Mapping + +from .store import payload_digest + + +FACT_TABLES = ("paper_orders", "paper_fills", "paper_positions", "paper_equity_curve", "paper_reconciliation") + + +def extract_daily_report_facts(report: Mapping[str, Any]) -> dict[str, tuple[Mapping[str, Any], ...]]: + session_id = str(report["session_id"]) + orders = tuple(_row(session_id, item, "order_id", index) for index, item in enumerate(_items(report.get("order_intents", report.get("orders", ()))))) + fills = tuple(_row(session_id, item, "fill_id", index) for index, item in enumerate(_items(report.get("fills", ())))) + ledger = _mapping(report.get("ledger_after")) + positions = tuple( + _row(session_id, {"symbol": symbol, "quantity": quantity}, "symbol", index) + for index, (symbol, quantity) in enumerate(sorted(_mapping(ledger.get("positions")).items())) + ) + equity_source = report.get("equity_curve", report.get("equity")) + equity = tuple(_row(session_id, item, "timestamp", index) for index, item in enumerate(_items(equity_source))) + if not equity and isinstance(equity_source, Mapping): + equity = (_row(session_id, equity_source, "session_id", 0),) + reconciliation_source = report.get("reconciliation_audit", report.get("reconciliation")) + reconciliation = tuple(_row(session_id, item, "reconciliation_id", index) for index, item in enumerate(_items(reconciliation_source))) + if not reconciliation and isinstance(reconciliation_source, Mapping): + reconciliation = (_row(session_id, reconciliation_source, "session_id", 0),) + return {"paper_orders": orders, "paper_fills": fills, "paper_positions": positions, "paper_equity_curve": equity, "paper_reconciliation": reconciliation} + + +def _items(value: Any) -> tuple[Mapping[str, Any], ...]: + if isinstance(value, Mapping): + return (dict(value),) + return tuple(dict(item) for item in (value or ()) if isinstance(item, Mapping)) + + +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _row(session_id: str, source: Mapping[str, Any], preferred_identity: str, index: int) -> Mapping[str, Any]: + payload = dict(source) + identity = payload.get(preferred_identity) or payload.get("order_id") or payload.get("fill_id") + if identity is None: + identity = payload.get("symbol") or payload.get("timestamp") or payload_digest(payload)[:16] + return {"session_id": session_id, "source_identity": str(identity), "source_index": index, "payload": payload, "payload_digest": payload_digest(payload)} diff --git a/src/quantpilot_core/continuous_paper/learning.py b/src/quantpilot_core/continuous_paper/learning.py new file mode 100644 index 0000000..24803d4 --- /dev/null +++ b/src/quantpilot_core/continuous_paper/learning.py @@ -0,0 +1,49 @@ +"""Read-only adapter from existing Quant Firm Learning Desk contracts.""" +from __future__ import annotations + +from dataclasses import asdict, is_dataclass +from typing import Any, Mapping + + +def build_learning_payload(report: Mapping[str, Any], session_id: str, shadow_delta: Mapping[str, Any] | None = None) -> Mapping[str, Any]: + quant_firm = _mapping(report.get("quant_firm")) + desk = _mapping(quant_firm.get("learning_desk")) or _mapping(report.get("learning_desk")) + mutation = _mapping(desk.get("strategy_mutation")) or _mapping(desk.get("strategy_mutation_plan")) + recommendations = mutation.get("recommendations", ()) + proposals = tuple(_proposal(item, session_id) for item in recommendations if isinstance(item, Mapping)) + return { + "source_session_id": session_id, + "attribution": _json(desk.get("attribution", {})), + "experiment": _json(desk.get("experiment", desk.get("experiment_record", {}))), + "failure_analysis": _json(desk.get("failure_analysis", {})), + "strategy_mutation": _json(mutation), + "parameter_proposals": proposals, + "evidence_refs": tuple(desk.get("evidence_refs", ())) or (f"paper-session://{session_id}",), + "production_vs_shadow_delta": dict(shadow_delta or {}), + } + + +def _proposal(item: Mapping[str, Any], session_id: str) -> Mapping[str, Any]: + proposed = item.get("proposed_value", item.get("recommended_value", item.get("value"))) + current = item.get("current_value", item.get("baseline_value")) + return { + "source_session_id": session_id, + "parameter_name": item.get("parameter_name", item.get("parameter", "unknown")), + "current_value": current, + "proposed_value": proposed, + "rationale": item.get("rationale", item.get("reason", "")), + "supporting_evidence_refs": tuple(item.get("supporting_evidence_refs", item.get("evidence_refs", ()))), + "status": "proposed", + "requires_offline_validation": True, + } + + +def _mapping(value: Any) -> Mapping[str, Any]: + return value if isinstance(value, Mapping) else {} + + +def _json(value: Any) -> Any: + if is_dataclass(value) and not isinstance(value, type): return _json(asdict(value)) + if isinstance(value, Mapping): return {str(k): _json(v) for k, v in value.items()} + if isinstance(value, (tuple, list)): return [_json(v) for v in value] + return getattr(value, "value", value) diff --git a/src/quantpilot_core/continuous_paper/prefect_flow.py b/src/quantpilot_core/continuous_paper/prefect_flow.py new file mode 100644 index 0000000..706b625 --- /dev/null +++ b/src/quantpilot_core/continuous_paper/prefect_flow.py @@ -0,0 +1,39 @@ +"""Optional deployable Prefect wrapper; imports have no scheduling side effects.""" +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any + +from quantpilot_core.production_candidate import load_runtime_manifest +from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig +from .active_shadow import ActiveShadowConfig +from .contracts import ContinuousPaperCycleConfig +from .service import ContinuousPaperCycle +from .store import PostgreSQLReportingStore + +SHANGHAI_TIMEZONE = "Asia/Shanghai" + + +def retryable_exception(exc: BaseException) -> bool: + if exc.__class__.__name__ in {"ReportingConflictError", "ValueError", "TypeError"}: + return False + return any(word in str(exc).lower() for word in ("connection", "timeout", "temporar", "postgres", "network")) + + +def run_continuous_paper_flow(*, production_manifest_path: str, decision_session: str, state_path: str, report_path: str, run_id: str, dsn_env_var: str = "QUANTPILOT_POSTGRES_DSN", active_shadow: dict[str, Any] | None = None, **pipeline_options: Any) -> Any: + """Construct all runtime objects from serializable scheduled-flow parameters.""" + dsn = os.environ.get(dsn_env_var) + if not dsn: + raise RuntimeError(f"required PostgreSQL DSN environment variable is not set: {dsn_env_var}") + manifest = load_runtime_manifest(Path(production_manifest_path)) + pipeline = RealCandidatePipelineConfig(decision_session=decision_session, state_path=state_path, report_path=report_path, production_manifest=manifest, **pipeline_options) + config = ContinuousPaperCycleConfig(pipeline_config=pipeline, run_id=run_id, active_shadow=ActiveShadowConfig(**(active_shadow or {}))) + return ContinuousPaperCycle(PostgreSQLReportingStore(dsn)).run_once(config) + + +try: + from prefect import flow + continuous_paper_flow = flow(name="quantpilot-continuous-paper-v1", retries=2, retry_condition_fn=lambda _task, _state, _run: retryable_exception(_state.result(raise_on_failure=False)))(run_continuous_paper_flow) +except ImportError: + continuous_paper_flow = run_continuous_paper_flow diff --git a/src/quantpilot_core/continuous_paper/service.py b/src/quantpilot_core/continuous_paper/service.py new file mode 100644 index 0000000..d7e792a --- /dev/null +++ b/src/quantpilot_core/continuous_paper/service.py @@ -0,0 +1,67 @@ +"""Application service coordinating exactly one existing production paper cycle.""" +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Callable, Mapping + +from quantpilot_core.production_candidate import load_runtime_manifest, manifest_payload +from quantpilot_core.real_candidate_pipeline import run_real_candidate_daily_paper + +from .active_shadow import ActiveShadowConfig, ActiveShadowRunner +from .contracts import ContinuousPaperCycleConfig, ContinuousPaperCycleResult, ReportingCycleBundle +from .facts import extract_daily_report_facts +from .learning import build_learning_payload +from .store import InMemoryReportingStore, ReportingStore, payload_digest + + +class ContinuousPaperCycle: + def __init__(self, store: ReportingStore | None = None, *, pipeline_runner: Callable[..., Any] = run_real_candidate_daily_paper, shadow_runner_factory: Callable[[ActiveShadowConfig], Any] = ActiveShadowRunner, clock: Callable[[], datetime] | None = None) -> None: + self.store = store or InMemoryReportingStore() + self.pipeline_runner = pipeline_runner + self.shadow_runner_factory = shadow_runner_factory + self.clock = clock or (lambda: datetime.now(timezone.utc)) + + def run_once(self, config: ContinuousPaperCycleConfig, **pipeline_dependencies: Any) -> ContinuousPaperCycleResult: + self.store.initialize() + # Validate strict manifest binding before production work begins. + manifest = load_runtime_manifest(config.pipeline_config.production_manifest) + started_at = self.clock() + pipeline_result = self.pipeline_runner(config.pipeline_config, **pipeline_dependencies) + daily = pipeline_result.daily_paper_loop_result + if daily is None: + raise RuntimeError("real_candidate_pipeline did not produce a daily paper result") + report = dict(daily.report) + session_id = str(report.get("session_id", "")) + if not session_id: + raise ValueError("daily paper report must contain session_id") + report_digest = payload_digest(report) + production_input_digest = str(report.get("input_digest", "")) + completed_at = self.clock() + shadow_config = config.active_shadow if isinstance(config.active_shadow, ActiveShadowConfig) else ActiveShadowConfig() + provisional_learning = build_learning_payload(report, session_id) + try: + shadow = self.shadow_runner_factory(shadow_config).run({"session_id": session_id, "daily_report": report, "learning_desk": provisional_learning}) + except Exception as exc: # Defensive boundary: advisory never loses production facts. + shadow = {"mode": "abstain", "evidence_digest": payload_digest(report), "physical_model_calls": 0, "cache_hits": 0, "estimated_api_cost": 0.0, "abstentions": 1, "reason": f"shadow_runner_failure:{type(exc).__name__}"} + delta = _production_shadow_delta(report, shadow) + learning = build_learning_payload(report, session_id, delta) + session = { + "session_id": session_id, "run_id": config.run_id, "decision_session": report["decision_session"], + "execution_session": report.get("execution_session"), "status": daily.status.value, + "strategy_id": config.pipeline_config.strategy_id, + "production_manifest_digest": payload_digest(manifest_payload(manifest)), + "effective_parameter_digest": report.get("effective_parameter_digest"), + "production_input_digest": production_input_digest, + "state_hash_before": report.get("state_hash_before"), "state_hash_after": report.get("state_hash_after"), + "prefect_flow_id": config.prefect_flow_id, "prefect_deployment_id": config.prefect_deployment_id, + "prefect_flow_run_id": config.prefect_flow_run_id, "replay_status": daily.status.value, + "failure_class": None, "failure_reason": None, "started_at": started_at, "completed_at": completed_at, + "report_digest": report_digest, + } + self.store.persist_cycle(ReportingCycleBundle(session, report, learning, shadow, extract_daily_report_facts(report))) + return ContinuousPaperCycleResult(session_id, config.run_id, daily.status.value, report_digest, production_input_digest, started_at, completed_at, daily.status.value, shadow) + + +def _production_shadow_delta(report: Mapping[str, Any], shadow: Mapping[str, Any]) -> Mapping[str, Any]: + """A reporting comparison only; does not feed any execution artifact.""" + return {"production_report_digest": payload_digest(report), "shadow_digest": payload_digest(shadow), "shadow_mode": shadow.get("mode"), "shadow_abstentions": shadow.get("abstentions", 0)} diff --git a/src/quantpilot_core/continuous_paper/store.py b/src/quantpilot_core/continuous_paper/store.py new file mode 100644 index 0000000..76edcf5 --- /dev/null +++ b/src/quantpilot_core/continuous_paper/store.py @@ -0,0 +1,153 @@ +"""Atomic reporting persistence; it stores source facts and calculates nothing.""" +from __future__ import annotations + +import hashlib +import json +from contextlib import contextmanager +from typing import Any, Mapping, Protocol + + +SCHEMA_VERSION = 2 + + +def canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + + +def payload_digest(value: Any) -> str: + return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +class ReportingConflictError(ValueError): + """A stable reporting identity was replayed with different immutable data.""" + + +class ReportingStore(Protocol): + def initialize(self) -> None: ... + def persist_cycle(self, bundle: Any) -> None: ... + def get_session(self, session_id: str) -> Mapping[str, Any] | None: ... + + +class InMemoryReportingStore: + def __init__(self) -> None: + self.sessions: dict[str, dict[str, Any]] = {} + self.reports: dict[str, dict[str, Any]] = {} + self.learning: dict[str, dict[str, Any]] = {} + self.shadow: dict[str, dict[str, Any]] = {} + self.facts: dict[str, dict[tuple[str, str, int], dict[str, Any]]] = {name: {} for name in ("paper_orders", "paper_fills", "paper_positions", "paper_equity_curve", "paper_reconciliation")} + + def initialize(self) -> None: + return None + + def persist_cycle(self, bundle: Any) -> None: + # Validate against copies first: failure is all-or-nothing. + tables = (dict(self.sessions), dict(self.reports), dict(self.learning), dict(self.shadow), {name: dict(rows) for name, rows in self.facts.items()}) + self._write_bundle(tables, bundle) + self.sessions, self.reports, self.learning, self.shadow, self.facts = tables + + def _write_bundle(self, tables: Any, bundle: Any) -> None: + sessions, reports, learning, shadow, facts = tables + session = dict(bundle.session); sid = str(session["session_id"]) + self._immutable(sessions, sid, session, "report_digest") + self._immutable(reports, sid, {"payload": dict(bundle.report)}, "payload_digest") + self._immutable(learning, sid, {"payload": dict(bundle.learning)}, "payload_digest") + self._immutable(shadow, sid, {"payload": dict(bundle.shadow)}, "payload_digest") + for table, rows in bundle.facts.items(): + if table not in facts: + raise ValueError(f"unsupported fact table: {table}") + for row in rows: + item = dict(row) + key = (sid, str(item["source_identity"]), int(item.get("source_index", 0))) + self._immutable(facts[table], key, item, "payload_digest") + + @staticmethod + def _immutable(table: dict[Any, dict[str, Any]], key: Any, value: Mapping[str, Any], digest_key: str) -> None: + row = dict(value) + row.setdefault(digest_key, payload_digest(row.get("payload", row))) + prior = table.get(key) + if prior is not None and prior[digest_key] != row[digest_key]: + raise ReportingConflictError(f"conflicting digest for identity: {key}") + table.setdefault(key, row) + + # Legacy helpers retained as small adapters for callers outside this PR. + def upsert_session(self, session: Mapping[str, Any]) -> None: self._immutable(self.sessions, str(session["session_id"]), session, "report_digest") + def upsert_report(self, sid: str, payload: Mapping[str, Any]) -> None: self._immutable(self.reports, sid, {"payload": dict(payload)}, "payload_digest") + def upsert_learning(self, sid: str, payload: Mapping[str, Any]) -> None: self._immutable(self.learning, sid, {"payload": dict(payload)}, "payload_digest") + def upsert_shadow(self, sid: str, payload: Mapping[str, Any]) -> None: self._immutable(self.shadow, sid, {"payload": dict(payload)}, "payload_digest") + def get_session(self, session_id: str) -> Mapping[str, Any] | None: return self.sessions.get(session_id) + + +SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS reporting_schema_version (version INTEGER PRIMARY KEY); +INSERT INTO reporting_schema_version(version) VALUES (2) ON CONFLICT DO NOTHING; +CREATE TABLE IF NOT EXISTS paper_sessions (session_id TEXT PRIMARY KEY, run_id TEXT NOT NULL, decision_session TEXT NOT NULL, execution_session TEXT, status TEXT NOT NULL, strategy_id TEXT, production_manifest_digest TEXT, effective_parameter_digest TEXT, production_input_digest TEXT, state_hash_before TEXT, state_hash_after TEXT, prefect_flow_id TEXT, prefect_deployment_id TEXT, prefect_flow_run_id TEXT, replay_status TEXT NOT NULL, failure_class TEXT, failure_reason TEXT, started_at TIMESTAMPTZ NOT NULL, completed_at TIMESTAMPTZ NOT NULL, report_digest TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS paper_session_reports (session_id TEXT PRIMARY KEY REFERENCES paper_sessions(session_id), schema_version INTEGER NOT NULL, payload JSONB NOT NULL, payload_digest TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS learning_desk_reports (session_id TEXT PRIMARY KEY REFERENCES paper_sessions(session_id), payload JSONB NOT NULL, payload_digest TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS ai_shadow_reports (session_id TEXT PRIMARY KEY REFERENCES paper_sessions(session_id), mode TEXT NOT NULL, payload JSONB NOT NULL, payload_digest TEXT NOT NULL, evidence_digest TEXT NOT NULL, physical_model_calls INTEGER NOT NULL, cache_hits INTEGER NOT NULL, estimated_api_cost DOUBLE PRECISION NOT NULL); +CREATE TABLE IF NOT EXISTS paper_orders (session_id TEXT REFERENCES paper_sessions(session_id), source_identity TEXT NOT NULL, source_index INTEGER NOT NULL, payload JSONB NOT NULL, payload_digest TEXT NOT NULL, PRIMARY KEY(session_id,source_identity,source_index)); +CREATE TABLE IF NOT EXISTS paper_fills (session_id TEXT REFERENCES paper_sessions(session_id), source_identity TEXT NOT NULL, source_index INTEGER NOT NULL, payload JSONB NOT NULL, payload_digest TEXT NOT NULL, PRIMARY KEY(session_id,source_identity,source_index)); +CREATE TABLE IF NOT EXISTS paper_positions (session_id TEXT REFERENCES paper_sessions(session_id), source_identity TEXT NOT NULL, source_index INTEGER NOT NULL, payload JSONB NOT NULL, payload_digest TEXT NOT NULL, PRIMARY KEY(session_id,source_identity,source_index)); +CREATE TABLE IF NOT EXISTS paper_equity_curve (session_id TEXT REFERENCES paper_sessions(session_id), source_identity TEXT NOT NULL, source_index INTEGER NOT NULL, payload JSONB NOT NULL, payload_digest TEXT NOT NULL, PRIMARY KEY(session_id,source_identity,source_index)); +CREATE TABLE IF NOT EXISTS paper_reconciliation (session_id TEXT REFERENCES paper_sessions(session_id), source_identity TEXT NOT NULL, source_index INTEGER NOT NULL, payload JSONB NOT NULL, payload_digest TEXT NOT NULL, PRIMARY KEY(session_id,source_identity,source_index)); +""" + + +class PostgreSQLReportingStore: + def __init__(self, dsn: str | None = None, *, connection: Any | None = None) -> None: + self.dsn, self.connection = dsn, connection + + def _conn(self) -> Any: + if self.connection is None: + try: + import psycopg + except ImportError as exc: + raise RuntimeError("install quantpilot-ai-next[continuous-paper] for PostgreSQL") from exc + self.connection = psycopg.connect(self.dsn) + return self.connection + + def initialize(self) -> None: + conn = self._conn() + with conn.cursor() as cur: + cur.execute(SCHEMA_SQL) + conn.commit() + + def persist_cycle(self, bundle: Any) -> None: + conn = self._conn() + try: + with conn.cursor() as cur: + self._check_and_insert(cur, "paper_sessions", "session_id", dict(bundle.session), "report_digest", session=True) + self._check_and_insert(cur, "paper_session_reports", "session_id", {"session_id": bundle.session["session_id"], "schema_version": SCHEMA_VERSION, "payload": bundle.report}, "payload_digest") + self._check_and_insert(cur, "learning_desk_reports", "session_id", {"session_id": bundle.session["session_id"], "payload": bundle.learning}, "payload_digest") + shadow = dict(bundle.shadow) + self._check_and_insert(cur, "ai_shadow_reports", "session_id", {"session_id": bundle.session["session_id"], "mode": shadow.get("mode", "disabled"), "payload": shadow, "evidence_digest": shadow.get("evidence_digest", ""), "physical_model_calls": shadow.get("physical_model_calls", 0), "cache_hits": shadow.get("cache_hits", 0), "estimated_api_cost": shadow.get("estimated_api_cost", 0.0)}, "payload_digest") + for table, rows in bundle.facts.items(): + for row in rows: + self._check_and_insert(cur, table, "session_id,source_identity,source_index", dict(row), "payload_digest") + conn.commit() + except Exception: + conn.rollback() + raise + + def _check_and_insert(self, cur: Any, table: str, keys: str, row: Mapping[str, Any], digest_key: str, *, session: bool = False) -> None: + item = dict(row); item.setdefault(digest_key, payload_digest(item.get("payload", item))) + key_names = keys.split(",") + where = " AND ".join(f"{name}=%s" for name in key_names) + args = tuple(item[name] for name in key_names) + cur.execute(f"SELECT {digest_key} FROM {table} WHERE {where}", args) + old = cur.fetchone() + if old: + if old[0] != item[digest_key]: + raise ReportingConflictError(f"conflicting digest for {table}: {args}") + return + columns = list(item) + values = [] + for name in columns: + values.append(canonical_json(item[name]) if name == "payload" else item[name]) + placeholders = ",".join("%s::jsonb" if name == "payload" else "%s" for name in columns) + cur.execute(f"INSERT INTO {table}({','.join(columns)}) VALUES({placeholders})", tuple(values)) + + def get_session(self, session_id: str) -> Mapping[str, Any] | None: + with self._conn().cursor() as cur: + cur.execute("SELECT session_id, report_digest FROM paper_sessions WHERE session_id=%s", (session_id,)) + row = cur.fetchone() + return {"session_id": row[0], "report_digest": row[1]} if row else None diff --git a/tests/continuous_paper/test_contracts_and_shadow.py b/tests/continuous_paper/test_contracts_and_shadow.py new file mode 100644 index 0000000..c3ef93c --- /dev/null +++ b/tests/continuous_paper/test_contracts_and_shadow.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import json + +import pytest + +from quantpilot_core.continuous_paper.active_shadow import ActiveShadowConfig, ActiveShadowRunner +from quantpilot_core.continuous_paper.contracts import ReportingCycleBundle +from quantpilot_core.continuous_paper.store import InMemoryReportingStore, ReportingConflictError + + +def _bundle(payload: dict[str, object]) -> ReportingCycleBundle: + return ReportingCycleBundle( + session={"session_id": "session-1", "run_id": "run-1", "report_digest": "report"}, + report=payload, + learning={}, shadow={"mode": "disabled"}, + facts={"paper_orders": (), "paper_fills": (), "paper_positions": (), "paper_equity_curve": (), "paper_reconciliation": ()}, + ) + + +def test_memory_store_replay_is_idempotent_and_conflict_safe() -> None: + store = InMemoryReportingStore() + store.persist_cycle(_bundle({"value": 1})) + store.persist_cycle(_bundle({"value": 1})) + with pytest.raises(ReportingConflictError): + store.persist_cycle(_bundle({"value": 2})) + + +def test_disabled_shadow_has_no_physical_requests(tmp_path) -> None: + result = ActiveShadowRunner(ActiveShadowConfig(cache_path=tmp_path)).run({"pit": "evidence"}) + assert result["mode"] == "disabled" + assert result["physical_model_calls"] == 0 + + +def test_malformed_cache_abstains_without_provider_call(tmp_path) -> None: + (tmp_path / "bad.json").write_text("not json") + result = ActiveShadowRunner(ActiveShadowConfig(enabled=True, cache_path=tmp_path)).run({"pit": "evidence"}) + assert result["physical_model_calls"] == 0 + assert result["abstentions"] > 0 diff --git a/tests/continuous_paper/test_prefect_and_active_shadow.py b/tests/continuous_paper/test_prefect_and_active_shadow.py new file mode 100644 index 0000000..7db92c1 --- /dev/null +++ b/tests/continuous_paper/test_prefect_and_active_shadow.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import importlib + +from quantpilot_core.continuous_paper.active_shadow import ActiveShadowConfig, ActiveShadowRunner +from quantpilot_core.quant_firm.deepseek_advisory import DeepSeekAdvisoryRole + + +def test_prefect_module_import_has_no_scheduling_side_effect(monkeypatch) -> None: + module = importlib.import_module("quantpilot_core.continuous_paper.prefect_flow") + assert module.SHANGHAI_TIMEZONE == "Asia/Shanghai" + assert module.retryable_exception(ValueError("manifest mismatch")) is False + assert module.retryable_exception(RuntimeError("connection reset")) is True + + +class _FailingAgent: + config = type("Config", (), {"model": "test-model"})() + def advise(self, _input): raise RuntimeError("provider failed") + + +class _RecordingAgent: + config = type("Config", (), {"model": "test-model"})() + def __init__(self): self.inputs = [] + def advise(self, advisory_input): + self.inputs.append(advisory_input) + return {"status": "advisory", "role": advisory_input.role.value} + + +def test_failed_provider_counts_physical_request_and_cost(tmp_path) -> None: + config = ActiveShadowConfig(enabled=True, enable_live_calls=True, max_physical_model_calls_per_cycle=1, max_estimated_cost_per_cycle=2.0, estimated_cost_per_call=1.0, cache_path=tmp_path, allow_roles=(DeepSeekAdvisoryRole.RESEARCH_DESK,)) + result = ActiveShadowRunner(config, agent=_FailingAgent()).run({"pit": "safe"}) + assert result["physical_model_calls"] == 1 + assert result["estimated_api_cost"] == 1.0 + assert result["abstentions"] == 1 + + +def test_cost_cap_prevents_physical_request(tmp_path) -> None: + config = ActiveShadowConfig(enabled=True, enable_live_calls=True, max_physical_model_calls_per_cycle=1, max_estimated_cost_per_cycle=0.5, estimated_cost_per_call=1.0, cache_path=tmp_path, allow_roles=(DeepSeekAdvisoryRole.RESEARCH_DESK,)) + result = ActiveShadowRunner(config, agent=_FailingAgent()).run({"pit": "safe"}) + assert result["physical_model_calls"] == 0 + assert result["roles"][DeepSeekAdvisoryRole.RESEARCH_DESK.value]["reason"] == "cost_budget_exhausted" + + +def test_committee_receives_bound_specialist_evidence_digests(tmp_path) -> None: + agent = _RecordingAgent() + roles = (DeepSeekAdvisoryRole.RESEARCH_DESK, DeepSeekAdvisoryRole.INVESTMENT_COMMITTEE) + config = ActiveShadowConfig(enabled=True, enable_live_calls=True, max_physical_model_calls_per_cycle=2, max_estimated_cost_per_cycle=2.0, estimated_cost_per_call=1.0, cache_path=tmp_path, allow_roles=roles) + ActiveShadowRunner(config, agent=agent).run({"pit": "safe"}) + committee_input = agent.inputs[-1].quant_firm_decision_report_summary + assert committee_input["specialist_evidence_digests"] diff --git a/tests/continuous_paper/test_service_learning_store.py b/tests/continuous_paper/test_service_learning_store.py new file mode 100644 index 0000000..5733ed6 --- /dev/null +++ b/tests/continuous_paper/test_service_learning_store.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +import quantpilot_core.continuous_paper.service as service_module +from quantpilot_core.continuous_paper.active_shadow import ActiveShadowConfig +from quantpilot_core.continuous_paper.contracts import ContinuousPaperCycleConfig, ReportingCycleBundle +from quantpilot_core.continuous_paper.learning import build_learning_payload +from quantpilot_core.continuous_paper.service import ContinuousPaperCycle +from quantpilot_core.continuous_paper.store import InMemoryReportingStore, ReportingConflictError +from quantpilot_core.real_candidate_pipeline import RealCandidatePipelineConfig + + +def _report() -> dict[str, object]: + return { + "session_id": "paper-20260712", "decision_session": "2026-07-12", "execution_session": "2026-07-13", + "input_digest": "production-input", "order_intents": [{"order_id": "order-1", "symbol": "600000.SH"}], + "fills": [{"fill_id": "fill-1", "symbol": "600000.SH"}], + "ledger_after": {"positions": {"600000.SH": 100}, "cash": 900.0}, + "reconciliation_audit": {"reconciliation_id": "reconcile-1", "status": "ok"}, + "quant_firm": {"learning_desk": {"strategy_mutation": {"recommendations": [{"parameter_name": "risk_penalty", "current_value": 1.0, "proposed_value": 1.25, "rationale": "drawdown evidence", "evidence_refs": ["evidence://drawdown"]}]}}}, + } + + +class _ShadowFailure: + def __init__(self, _config): pass + def run(self, _evidence): raise RuntimeError("provider unavailable") + + +class _ShadowSuccess: + def __init__(self, _config): pass + def run(self, _evidence): return {"mode": "active_shadow", "physical_model_calls": 0, "cache_hits": 0, "estimated_api_cost": 0.0, "abstentions": 0} + + +def _run(monkeypatch, shadow_factory): + calls = [] + report = _report() + monkeypatch.setattr(service_module, "load_runtime_manifest", lambda _manifest: {"manifest": "fixed"}) + monkeypatch.setattr(service_module, "manifest_payload", lambda manifest: manifest) + def pipeline(config, **kwargs): + calls.append((config, kwargs)) + return SimpleNamespace(daily_paper_loop_result=SimpleNamespace(report=report, status=SimpleNamespace(value="completed"))) + config = ContinuousPaperCycleConfig(RealCandidatePipelineConfig(decision_session="2026-07-12", production_manifest=object()), run_id="run-1", active_shadow=ActiveShadowConfig()) + store = InMemoryReportingStore() + result = ContinuousPaperCycle(store, pipeline_runner=pipeline, shadow_runner_factory=shadow_factory).run_once(config) + return result, store, calls + + +def test_service_calls_production_pipeline_once_and_shadow_failure_is_advisory(monkeypatch) -> None: + result, store, calls = _run(monkeypatch, _ShadowFailure) + assert len(calls) == 1 + assert result.status == "completed" + assert result.shadow_report["mode"] == "abstain" + assert store.sessions[result.session_id]["production_input_digest"] == "production-input" + + +def test_shadow_does_not_change_production_session_or_execution_facts(monkeypatch) -> None: + success, success_store, _ = _run(monkeypatch, _ShadowSuccess) + failed, failed_store, _ = _run(monkeypatch, _ShadowFailure) + assert (success.session_id, success.production_input_digest, success.report_digest) == (failed.session_id, failed.production_input_digest, failed.report_digest) + for table in ("paper_orders", "paper_fills", "paper_positions", "paper_reconciliation"): + assert success_store.facts[table] == failed_store.facts[table] + + +def test_learning_proposals_use_real_strategy_mutation_fields() -> None: + proposal = build_learning_payload(_report(), "paper-20260712")["parameter_proposals"][0] + assert proposal == { + "source_session_id": "paper-20260712", "parameter_name": "risk_penalty", "current_value": 1.0, + "proposed_value": 1.25, "rationale": "drawdown evidence", "supporting_evidence_refs": ("evidence://drawdown",), + "status": "proposed", "requires_offline_validation": True, + } + + +def _bundle(*, session_digest="session", shadow=None, facts=None) -> ReportingCycleBundle: + return ReportingCycleBundle( + session={"session_id": "session-1", "run_id": "run-1", "report_digest": session_digest}, report={"report": 1}, learning={}, + shadow=shadow or {"mode": "disabled"}, facts=facts or {"paper_orders": (), "paper_fills": (), "paper_positions": (), "paper_equity_curve": (), "paper_reconciliation": ()}, + ) + + +def test_reporting_store_rolls_back_invalid_cycle_and_detail_rows_are_idempotent() -> None: + store = InMemoryReportingStore() + facts = {"paper_orders": ({"session_id": "session-1", "source_identity": "order-1", "source_index": 0, "payload": {"order_id": "order-1"}},), "paper_fills": (), "paper_positions": (), "paper_equity_curve": (), "paper_reconciliation": ()} + store.persist_cycle(_bundle(facts=facts)); store.persist_cycle(_bundle(facts=facts)) + assert len(store.facts["paper_orders"]) == 1 + with pytest.raises(ValueError): store.persist_cycle(ReportingCycleBundle({"session_id": "bad", "run_id": "run", "report_digest": "x"}, {}, {}, {}, {"not_a_table": ()})) + assert "bad" not in store.sessions + + +def test_reporting_store_rejects_conflicting_session_and_shadow_digests() -> None: + store = InMemoryReportingStore(); store.persist_cycle(_bundle()) + with pytest.raises(ReportingConflictError): store.persist_cycle(_bundle(session_digest="other")) + with pytest.raises(ReportingConflictError): store.persist_cycle(_bundle(shadow={"mode": "active_shadow", "evidence_digest": "other"}))