PR #124: Add continuous paper operation and Control Center

infra/control_center/.env.example

@@ -0,0 +1 @@
+POSTGRES_PASSWORD=replace_me

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.

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: {}}

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"}]}]}

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"}]}]}

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"}]}]}

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"}]}]}

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"}]}]}

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}

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}

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"]

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()

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()

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",
+]

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)

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], ...]]
+