malkhut(spec): item 4 — ScenarioLibrary sweep for Mode 1 coverage
ScenarioLibrary sweeps the state space (not samples) across: - spread_mult: [0.1, 0.5, 1.0, 2.0, 5.0, 10.0] - depth_fraction: [0.01, 0.05, 0.1, 0.3, 0.5, 1.0] - toxicity: [0.0, 0.3, 0.7, 1.0] - regime: [normal, crisis, recovery, transition] Default: 13 assets × 576 grid points = 7,488 scenarios. Customizable: specify symbols, dimensions, ranges. 7 tests covering: grid size, sweep output, point fields, regime coverage, custom dimensions, summary, factory function.
This commit is contained in:
97
MALKHUT/malkhut/training/scenario_library.py
Normal file
97
MALKHUT/malkhut/training/scenario_library.py
Normal file
@@ -0,0 +1,97 @@
|
||||
"""
|
||||
ScenarioLibrary — sweeps the state space for comprehensive Mode 1 coverage.
|
||||
|
||||
Unlike ScenarioFactory which builds 30 pre-defined scenario types per asset,
|
||||
ScenarioLibrary SWEEPS across parameter dimensions to ensure coverage of
|
||||
regions the tape never visited.
|
||||
|
||||
Sweep dimensions (orthogonal):
|
||||
- spread_bps: [0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
|
||||
- depth_fraction: [0.01, 0.05, 0.1, 0.3, 0.5, 1.0]
|
||||
- toxicity: [0.0, 0.3, 0.7, 1.0]
|
||||
- regime: ["normal", "crisis", "recovery"]
|
||||
|
||||
Total: 6 × 6 × 4 × 4 = 576 grid points per asset.
|
||||
With 3 assets = 1,728 scenarios total.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, List, Optional, Sequence, Tuple
|
||||
|
||||
from malkhut.training.asset_classification import ASSET_PROFILES, get_asset_profile
|
||||
|
||||
|
||||
# Sweep dimension defaults
|
||||
DEFAULT_SPREAD_MULTS = [0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
|
||||
DEFAULT_DEPTH_FRACS = [0.01, 0.05, 0.1, 0.3, 0.5, 1.0]
|
||||
DEFAULT_TOXICITY = [0.0, 0.3, 0.7, 1.0]
|
||||
DEFAULT_REGIMES = ["normal", "crisis", "recovery", "transition"]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SweepPoint:
|
||||
"""A single point in the swept state space."""
|
||||
symbol: str
|
||||
spread_mult: float
|
||||
depth_fraction: float
|
||||
toxicity: float
|
||||
regime: str
|
||||
label: str # human-readable: "BTC_spread2x_depth30pct_tox0.3_crisis"
|
||||
|
||||
|
||||
class ScenarioLibrary:
|
||||
"""Sweeps the state space for Mode 1 (EXPLORE) coverage.
|
||||
|
||||
Produces grid points across spread × depth × toxicity × regime.
|
||||
These are ANCHORS that the ecology fills between and beyond.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
symbols: Optional[Sequence[str]] = None,
|
||||
spread_mults: Optional[Sequence[float]] = None,
|
||||
depth_fracs: Optional[Sequence[float]] = None,
|
||||
tox_levels: Optional[Sequence[float]] = None,
|
||||
regimes: Optional[Sequence[str]] = None,
|
||||
) -> None:
|
||||
self.symbols = list(symbols or ASSET_PROFILES.keys())
|
||||
self.spread_mults = list(spread_mults or DEFAULT_SPREAD_MULTS)
|
||||
self.depth_fracs = list(depth_fracs or DEFAULT_DEPTH_FRACS)
|
||||
self.tox_levels = list(tox_levels or DEFAULT_TOXICITY)
|
||||
self.regimes = list(regimes or DEFAULT_REGIMES)
|
||||
|
||||
def sweep(self) -> List[SweepPoint]:
|
||||
"""Generate full grid sweep across all dimensions."""
|
||||
points = []
|
||||
for symbol in self.symbols:
|
||||
for spread in self.spread_mults:
|
||||
for depth in self.depth_fracs:
|
||||
for tox in self.tox_levels:
|
||||
for regime in self.regimes:
|
||||
label = f"{symbol}_s{spread}_d{depth}_t{tox}_r{regime}"
|
||||
points.append(SweepPoint(
|
||||
symbol=symbol,
|
||||
spread_mult=spread,
|
||||
depth_fraction=depth,
|
||||
toxicity=tox,
|
||||
regime=regime,
|
||||
label=label,
|
||||
))
|
||||
return points
|
||||
|
||||
def grid_size(self) -> int:
|
||||
return len(self.symbols) * len(self.spread_mults) * len(self.depth_fracs) * len(self.tox_levels) * len(self.regimes)
|
||||
|
||||
def summary(self) -> str:
|
||||
return (f"ScenarioLibrary: {len(self.symbols)} assets × "
|
||||
f"{len(self.spread_mults)} spreads × {len(self.depth_fracs)} depths × "
|
||||
f"{len(self.tox_levels)} tox × {len(self.regimes)} regimes = "
|
||||
f"{self.grid_size()} grid points")
|
||||
|
||||
|
||||
def get_scenario_library(
|
||||
symbols: Optional[Sequence[str]] = None,
|
||||
) -> ScenarioLibrary:
|
||||
"""Get default scenario library for given symbols."""
|
||||
return ScenarioLibrary(symbols=symbols)
|
||||
Reference in New Issue
Block a user