266 lines
18 KiB
Python
266 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import asdict
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from django.db import transaction
|
|
|
|
from control_plane.events.bus import EventBus
|
|
from control_plane.projects.models import Project, ProjectStatus
|
|
from control_plane.trading_studio.models import (
|
|
AllocationTier, BacktestRun, DataKind, EvidenceStatus, ExperimentStatus,
|
|
FailureType, FeatureDefinition, FeatureSetVersion, LiveStrategyRun,
|
|
MarketDataset, MarketDatasetVersion, RunStatus, ShadowRun, SplitKind,
|
|
QualificationReplayLedger, QualificationReplayRun, Strategy, StrategyCapitalAllocation, StrategyEvaluation, StrategyExperiment,
|
|
StrategyPromotionDecision, StrategyStage, StrategyVersion, TradingCohort,
|
|
TradingProject, TradingProjectStatus, TradingResearchReport,
|
|
)
|
|
from control_plane.trading_studio.profiles import CryptoTradingProfile, ExistingSystemProfile, HyperScalperProfile, TradingProjectProfile
|
|
from control_plane.trading_studio.qualification import Bar, Calibration, QualificationReplayV1, ScenarioSpec, ledger_payload
|
|
|
|
|
|
class TradingStudioService:
|
|
"""Canonical, offline-first trading research service.
|
|
|
|
This service intentionally has no exchange adapter. A future live stage must
|
|
be separately authorized and supplied with a deterministic risk executor.
|
|
"""
|
|
|
|
def __init__(self, *, profile: TradingProjectProfile | None = None, existing_system: ExistingSystemProfile | None = None, bus: EventBus | None = None) -> None:
|
|
self.profile = profile or CryptoTradingProfile()
|
|
self.existing_system = existing_system or HyperScalperProfile()
|
|
self.bus = bus or EventBus()
|
|
|
|
def import_hyperscalper(self, *, repository_path: str, slug: str = "crypto-hyperscalper") -> TradingProject:
|
|
project, _ = Project.objects.get_or_create(
|
|
name="Crypto Trading Studio", project_type="TRADING",
|
|
defaults={"goal": "Falsify trading hypotheses before any capital allocation.", "repository_path": repository_path, "status": ProjectStatus.ARCHAEOLOGY},
|
|
)
|
|
trading_project, _ = TradingProject.objects.update_or_create(
|
|
slug=slug,
|
|
defaults={
|
|
"project": project, "name": "Crypto / Hyperliquid Strategy Lab",
|
|
"goal": "Prefer no strategy to an overfit strategy.",
|
|
"repository_path": repository_path, "profile_name": self.profile.name,
|
|
"existing_system_profile": self.existing_system.name,
|
|
"status": TradingProjectStatus.ARCHAEOLOGY,
|
|
"metadata": {"live_execution_enabled": False, "maximum_automatic_live_capital": "0", "safety_note": "Trading Studio V0.1 is offline-only. No exchange calls are implemented."},
|
|
},
|
|
)
|
|
self._event("TRADING_PROJECT_IMPORTED", trading_project, {"repository_path": repository_path})
|
|
return trading_project
|
|
|
|
def register_market_dataset(self, trading_project: TradingProject, *, name: str, kind: str, version: str, reference: str, content_hash: str, fields: list[str], record_count: int, start_at: datetime | None, end_at: datetime | None, resolution: str, quality: dict[str, Any], temporal_splits: dict[str, Any]) -> MarketDatasetVersion:
|
|
if kind not in DataKind.values:
|
|
raise ValueError("Unknown market data kind.")
|
|
self._validate_temporal_splits(temporal_splits)
|
|
dataset, _ = MarketDataset.objects.get_or_create(trading_project=trading_project, name=name, defaults={"kind": kind})
|
|
if dataset.kind != kind:
|
|
raise ValueError("Market dataset kind cannot change after registration.")
|
|
return MarketDatasetVersion.objects.create(
|
|
dataset=dataset, version=version, reference=reference, content_hash=content_hash,
|
|
fields=fields, record_count=record_count, start_at=start_at, end_at=end_at,
|
|
resolution=resolution, quality=quality, temporal_splits=temporal_splits,
|
|
lookahead_risk_status=EvidenceStatus.UNKNOWN, quality_status=EvidenceStatus.SUSPICIOUS,
|
|
)
|
|
|
|
def import_features(self, trading_project: TradingProject) -> list[FeatureDefinition]:
|
|
features = []
|
|
for item in self.existing_system.feature_inventory(trading_project.repository_path):
|
|
feature, _ = FeatureDefinition.objects.get_or_create(
|
|
trading_project=trading_project, name=item["name"], code_hash=item["code_hash"],
|
|
defaults={"family": item["family"], "implementation_reference": item["implementation_reference"], "leakage_status": EvidenceStatus.UNKNOWN},
|
|
)
|
|
features.append(feature)
|
|
return features
|
|
|
|
def create_feature_set(self, trading_project: TradingProject, *, name: str, version: str, features: list[FeatureDefinition]) -> FeatureSetVersion:
|
|
if any(feature.trading_project_id != trading_project.id for feature in features):
|
|
raise ValueError("Feature sets cannot cross trading projects.")
|
|
content_hash = self._hash({"features": sorted(str(feature.id) for feature in features)})
|
|
feature_set = FeatureSetVersion.objects.create(trading_project=trading_project, name=name, version=version, content_hash=content_hash)
|
|
feature_set.feature_definitions.set(features)
|
|
return feature_set
|
|
|
|
def create_strategy_version(self, trading_project: TradingProject, *, name: str, genome: dict[str, Any], feature_set: FeatureSetVersion | None = None, parent_version: StrategyVersion | None = None) -> StrategyVersion:
|
|
if self._contains_prohibited_sizing(genome):
|
|
raise ValueError("Martingale, loss chasing, and uncapped averaging are prohibited.")
|
|
if feature_set and feature_set.feature_definitions.filter(leakage_status=EvidenceStatus.BLOCKED).exists():
|
|
raise ValueError("Strategies using BLOCKED features cannot qualify.")
|
|
fingerprint = self._hash({"genome": genome, "feature_set": str(feature_set.id) if feature_set else ""})
|
|
strategy, _ = Strategy.objects.get_or_create(trading_project=trading_project, name=name)
|
|
return StrategyVersion.objects.create(strategy=strategy, version=f"v{strategy.versions.count() + 1}", parent_version=parent_version, feature_set=feature_set, genome=genome, fingerprint=fingerprint)
|
|
|
|
def propose_experiment(self, cohort: TradingCohort, strategy_version: StrategyVersion, contract: dict[str, Any]) -> StrategyExperiment:
|
|
required = ["hypothesis", "market_rationale", "expected_regime", "controls", "success_criteria", "rejection_criteria", "risk_assumptions", "execution_assumptions", "estimated_evaluation_cost"]
|
|
missing = [name for name in required if contract.get(name) in (None, "", {}, [])]
|
|
if missing:
|
|
raise ValueError("Incomplete strategy experiment contract: " + ", ".join(missing))
|
|
if strategy_version.strategy.trading_project_id != cohort.trading_project_id:
|
|
raise ValueError("Strategy and cohort must belong to the same TradingProject.")
|
|
if strategy_version.holdout_exposure_count and contract["controls"].get("uses_holdout_for_design"):
|
|
raise ValueError("Consumed holdout cannot be used for adaptive strategy design.")
|
|
return StrategyExperiment.objects.create(cohort=cohort, strategy_version=strategy_version, **{key: contract[key] for key in required})
|
|
|
|
@transaction.atomic
|
|
def record_backtest(self, experiment: StrategyExperiment, *, split: str, execution_model_version: str, metrics: dict[str, Any], configuration: dict[str, Any], artifact_reference: str = "") -> BacktestRun:
|
|
if split not in SplitKind.values:
|
|
raise ValueError("Unknown temporal split.")
|
|
if configuration.get("same_bar_close_execution"):
|
|
raise ValueError("Same-bar close execution is prohibited.")
|
|
required = ["gross_pnl", "fees", "funding", "slippage", "other_execution_cost", "net_pnl", "trade_count"]
|
|
missing = [name for name in required if name not in metrics]
|
|
if missing:
|
|
raise ValueError("Backtest metrics missing: " + ", ".join(missing))
|
|
computed_net = float(metrics["gross_pnl"]) - float(metrics["fees"]) - float(metrics["funding"]) - float(metrics["slippage"]) - float(metrics["other_execution_cost"])
|
|
if abs(computed_net - float(metrics["net_pnl"])) > 1e-9:
|
|
raise ValueError("Net PnL must equal gross PnL minus all execution costs.")
|
|
run = BacktestRun.objects.create(experiment=experiment, dataset_version=experiment.cohort.dataset_version, split=split, execution_model_version=execution_model_version, configuration=configuration, metrics=metrics, artifact_reference=artifact_reference, status=RunStatus.SUCCEEDED)
|
|
if split == SplitKind.HOLDOUT:
|
|
version = experiment.strategy_version
|
|
version.holdout_exposure_count += 1
|
|
version.immutable = True
|
|
version.save(update_fields=["holdout_exposure_count", "immutable", "updated_at"])
|
|
return run
|
|
|
|
@transaction.atomic
|
|
def run_qualification_canary(
|
|
self, *, strategy_version: StrategyVersion, dataset_version: MarketDatasetVersion, fold: str,
|
|
scenario: ScenarioSpec, runner_name: str, bars: list[Bar], calibration: Calibration,
|
|
protocol_identity: dict[str, Any] | None = None,
|
|
require_protocol_identity: bool = False,
|
|
) -> QualificationReplayRun:
|
|
"""Persist a supplied Cohort001 replay; never synthesize unavailable canary records."""
|
|
missing = [
|
|
name
|
|
for name, value in (
|
|
("strategy_version", strategy_version),
|
|
("dataset_version", dataset_version),
|
|
("fold", fold),
|
|
("bars", bars),
|
|
("calibration", calibration),
|
|
)
|
|
if not value
|
|
]
|
|
if missing:
|
|
raise ValueError(
|
|
"CANARY_ONLY refuses to run: missing materialized " + ", ".join(missing) + "."
|
|
)
|
|
result = QualificationReplayV1().run(
|
|
bars, strategy=str(strategy_version.id), dataset=str(dataset_version.id), fold=fold,
|
|
scenario=scenario, runner=runner_name, calibration=calibration,
|
|
)
|
|
configuration = self.qualification_configuration(
|
|
scenario,
|
|
protocol_identity=protocol_identity,
|
|
require_protocol_identity=require_protocol_identity,
|
|
)
|
|
run = QualificationReplayRun.objects.create(
|
|
strategy_version=strategy_version, dataset_version=dataset_version, fold=fold,
|
|
scenario_name=scenario.name, runner_name=runner_name, replay_mode=result.mode,
|
|
input_hash=calibration.input_hash, calibration=self._json_safe(asdict(calibration)),
|
|
configuration=configuration, summary={
|
|
"gross_pnl": result.gross_pnl, "net_pnl": result.net_pnl, "event_count": len(result.events),
|
|
"ledger_count": len(result.ledger), "rescue": self._json_safe(asdict(result.rescue)),
|
|
},
|
|
)
|
|
QualificationReplayLedger.objects.bulk_create([
|
|
QualificationReplayLedger(qualification_run=run, sequence=index, identity=row.identity,
|
|
result=self._json_safe(ledger_payload(row)))
|
|
for index, row in enumerate(result.ledger)
|
|
])
|
|
return run
|
|
|
|
@staticmethod
|
|
def qualification_configuration(
|
|
scenario: ScenarioSpec,
|
|
*,
|
|
protocol_identity: dict[str, Any] | None = None,
|
|
require_protocol_identity: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Build the immutable scenario snapshot used for persisted replay identity."""
|
|
if require_protocol_identity and not protocol_identity:
|
|
raise ValueError("Batch qualification requires a protocol identity snapshot.")
|
|
configuration = TradingStudioService._json_safe(asdict(scenario))
|
|
if protocol_identity is not None:
|
|
if not isinstance(protocol_identity, dict) or not protocol_identity:
|
|
raise ValueError("Protocol identity must be a non-empty object.")
|
|
configuration["protocol_identity"] = TradingStudioService._json_safe(protocol_identity)
|
|
return configuration
|
|
|
|
def judge_backtest(self, experiment: StrategyExperiment, backtest: BacktestRun, *, policy: dict[str, Any]) -> StrategyEvaluation:
|
|
metrics = backtest.metrics
|
|
failure = ""
|
|
verdict = "ADVANCE"
|
|
if float(metrics["net_pnl"]) < 0 <= float(metrics["gross_pnl"]):
|
|
failure, verdict = FailureType.FEE_DESTROYED, "KILL"
|
|
elif float(metrics["net_pnl"]) < 0:
|
|
failure, verdict = FailureType.NO_ALPHA, "KILL"
|
|
elif int(metrics["trade_count"]) < int(policy.get("minimum_trade_count", 0)):
|
|
failure, verdict = FailureType.INSUFFICIENT_TRADES, "REQUIRE_MORE_EVIDENCE"
|
|
elif float(metrics.get("pnl_concentration_top_trade", 0)) > float(policy.get("maximum_top_trade_concentration", 1)):
|
|
failure, verdict = FailureType.OVERFIT, "KILL"
|
|
evaluation = StrategyEvaluation.objects.create(strategy_version=experiment.strategy_version, experiment=experiment, stage=StrategyStage.BACKTEST, verdict=verdict, metrics=metrics, evidence={"policy": policy, "execution_model": backtest.execution_model_version}, failure_type=failure)
|
|
if verdict == "KILL":
|
|
experiment.status, experiment.conclusion, experiment.failure_type = ExperimentStatus.REJECTED, verdict, failure
|
|
experiment.strategy_version.stage = StrategyStage.KILLED
|
|
experiment.strategy_version.save(update_fields=["stage", "updated_at"])
|
|
else:
|
|
experiment.status, experiment.conclusion = ExperimentStatus.SUCCEEDED, verdict
|
|
experiment.save(update_fields=["status", "conclusion", "failure_type", "updated_at"])
|
|
return evaluation
|
|
|
|
def request_shadow(self, strategy_version: StrategyVersion) -> ShadowRun:
|
|
if strategy_version.stage not in {StrategyStage.HOLDOUT, StrategyStage.SHADOW}:
|
|
raise ValueError("Shadow requires a holdout-qualified immutable strategy version.")
|
|
return ShadowRun.objects.create(strategy_version=strategy_version, status=RunStatus.QUEUED, evidence={"mode": "PREPARED_ONLY", "live_execution_enabled": False})
|
|
|
|
def request_micro_live(self, strategy_version: StrategyVersion, amount: Decimal) -> LiveStrategyRun:
|
|
raise ValueError("Trading Studio V0.1 is offline-only. Explicit human authorization and a future deterministic execution/risk adapter are required before micro-live.")
|
|
|
|
def report(self, trading_project: TradingProject, *, title: str = "Trading cohort report") -> TradingResearchReport:
|
|
experiments = list(StrategyExperiment.objects.filter(cohort__trading_project=trading_project).select_related("strategy_version"))
|
|
payload = {"project": trading_project.slug, "status": trading_project.status, "offline_only": not trading_project.metadata.get("live_execution_enabled", False), "experiments": [{"strategy": item.strategy_version.strategy.name, "version": item.strategy_version.version, "status": item.status, "conclusion": item.conclusion, "failure_type": item.failure_type} for item in experiments], "counts": {"proposed": sum(item.status == ExperimentStatus.PROPOSED for item in experiments), "rejected": sum(item.status == ExperimentStatus.REJECTED for item in experiments), "survivors": sum(item.status == ExperimentStatus.SUCCEEDED for item in experiments)}}
|
|
return TradingResearchReport.objects.create(trading_project=trading_project, report_type="COHORT", title=title, content=payload, markdown="# TRADING COHORT REPORT\n\n" + json.dumps(payload, indent=2), evidence_status=EvidenceStatus.CONFIRMED)
|
|
|
|
def _validate_temporal_splits(self, splits: dict[str, Any]) -> None:
|
|
if splits.get("split_method") == "random":
|
|
raise ValueError("Random time-series splits are prohibited.")
|
|
ordered = [splits.get(kind) for kind in ["DISCOVERY", "TRAIN", "VALIDATION", "HOLDOUT", "FORWARD", "LIVE"] if splits.get(kind)]
|
|
previous_end = None
|
|
for item in ordered:
|
|
start, end = item.get("start"), item.get("end")
|
|
if not start or not end:
|
|
raise ValueError("Temporal split boundaries are required.")
|
|
if start >= end or previous_end and start < previous_end:
|
|
raise ValueError("Temporal splits must be chronological and non-overlapping.")
|
|
previous_end = end
|
|
|
|
def _contains_prohibited_sizing(self, value: Any) -> bool:
|
|
prohibited = ("martingale", "double_after_loss", "averaging_down", "loss_chasing", "negative_progression", "uncapped_grid")
|
|
if isinstance(value, dict):
|
|
return any(self._contains_prohibited_sizing(key) or self._contains_prohibited_sizing(item) for key, item in value.items())
|
|
if isinstance(value, list):
|
|
return any(self._contains_prohibited_sizing(item) for item in value)
|
|
return any(term in str(value).lower() for term in prohibited)
|
|
|
|
def _event(self, event_type: str, trading_project: TradingProject, payload: dict[str, Any]) -> None:
|
|
self.bus.publish(project=trading_project.project, event_type=event_type, payload=payload)
|
|
|
|
@staticmethod
|
|
def _hash(value: Any) -> str:
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()
|
|
|
|
@staticmethod
|
|
def _json_safe(value: Any) -> Any:
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
if isinstance(value, dict):
|
|
return {key: TradingStudioService._json_safe(item) for key, item in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [TradingStudioService._json_safe(item) for item in value]
|
|
return value
|