diff --git a/README.md b/README.md index 1c7a662..4a6f5ef 100644 --- a/README.md +++ b/README.md @@ -220,6 +220,7 @@ Record the current commit as the last-known-good SHA before deploy. If any gate - `python3 scripts/run_trading_plans.py` - generate today's structured BTC/ETH trading plans before signal generation; active plans require a Draw and at least one Invalidation - `python3 scripts/run_signals.py` - generate bias and signals - `python3 scripts/run_execution.py` - run paper execution gating, create an execution record, then auto-settle the latest execution for that signal +- `python3 scripts/run_risk_governor.py` - run the independent market-state and risk governor pre-execution gate - `python3 scripts/run_backtest.py` - run backtest and write results - `python3 scripts/run_autonomous_cycle.py` - run the unattended cycle: offline checks, database probe, readiness checks, then guarded pipeline when the database is available; prints `Next actions` for the next unattended step - `python3 scripts/run_database_readiness.py` - run read-only database readiness checks for connection, schema, seed data, and candle availability; prints `next_actions` and `action_plan` in JSON mode @@ -252,6 +253,7 @@ Optional utility JSON output variables: - `MIGRATION_CHECKS_OUTPUT_JSON` - `SCRIPT_CHECKS_OUTPUT_JSON` - `SHOW_MODELS_OUTPUT_JSON` +- `RISK_GOVERNOR_OUTPUT_JSON` - `TRADING_PLANS_OUTPUT_JSON` - `TRADING_REHEARSAL_OUTPUT_JSON` @@ -371,6 +373,20 @@ Optional account risk environment variables used by `run_execution.py` and `run_ - `EXECUTION_MAX_OPEN_POSITIONS` - `EXECUTION_MAX_SAME_SIDE_POSITIONS` +Optional market-state and risk governor environment variables used by `run_risk_governor.py`: + +- `RISK_GOVERNOR_OUTPUT_JSON` +- `RISK_INSTRUMENT_ID` +- `RISK_TIMEFRAME` +- `RISK_EXECUTION_MODE` +- `MARKET_STATE_MAX_CANDLE_AGE_SECONDS` +- `NEWS_WINDOW_ACTIVE` +- `NEWS_WINDOW_ID` +- `NEWS_WINDOW_MINUTES_BEFORE` +- `NEWS_WINDOW_MINUTES_AFTER` +- `OPENING_VOLATILITY_GUARD_MINUTES` +- `BLOCK_OPENING_VOLATILITY` + Optional signal quality environment variables used by `run_signals.py` and `run_pipeline.py`: - `SIGNALS_OUTPUT_JSON` - set to `1`, `true`, `yes`, or `on` for machine-readable `run_signals.py` output diff --git a/package.json b/package.json index 204305b..62cd49b 100644 --- a/package.json +++ b/package.json @@ -3,7 +3,7 @@ "version": "0.1.0", "private": true, "scripts": { - "lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service", + "lint": ".\\.venv\\Scripts\\python.exe scripts/run_script_static_checks.py && .\\.venv\\Scripts\\python.exe scripts/run_migration_static_checks.py && .\\.venv\\Scripts\\python.exe -m unittest tests.unit.test_trading_plan_service tests.unit.test_market_state_and_risk_governor tests.unit.test_execution_service tests.unit.test_api_server", "build": ".\\.venv\\Scripts\\python.exe -m compileall -q src scripts tests", "visual:snapshots": "node --check src/web/app.js" } diff --git a/scripts/run_risk_governor.py b/scripts/run_risk_governor.py new file mode 100644 index 0000000..573ccb0 --- /dev/null +++ b/scripts/run_risk_governor.py @@ -0,0 +1,105 @@ +from datetime import datetime, timedelta, timezone +from pathlib import Path +import json +import os +import sys + +from sqlalchemy.exc import SQLAlchemyError + +ROOT = Path(__file__).resolve().parents[1] +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from scripts.run_execution import build_account_risk_state_from_env +from src.services.market_state import DEFAULT_MAX_CANDLE_AGE_SECONDS, MarketStateService +from src.services.risk import RiskContext, RiskGovernorService + + +def main(env=None) -> int: + env = os.environ if env is None else env + output_json = _get_bool(env, "RISK_GOVERNOR_OUTPUT_JSON", False) + instrument_id = _get_int(env, "RISK_INSTRUMENT_ID", 1) + timeframe = str(env.get("RISK_TIMEFRAME") or "1m") + execution_mode = str(env.get("RISK_EXECUTION_MODE") or "paper") + service = MarketStateService() + try: + market_state = service.build_market_state( + instrument_id=instrument_id, + timeframe=timeframe, + max_candle_age_seconds=_get_int( + env, + "MARKET_STATE_MAX_CANDLE_AGE_SECONDS", + DEFAULT_MAX_CANDLE_AGE_SECONDS, + ), + news_windows=build_news_windows_from_env(env), + opening_volatility_minutes=_get_int(env, "OPENING_VOLATILITY_GUARD_MINUTES", 0), + ) + quality_gate = service.repository.fetch_latest_backtest_quality_gate(instrument_id=instrument_id) + except SQLAlchemyError as exc: + payload = { + "passed": False, + "error_type": exc.__class__.__name__, + "error": str(exc), + } + if output_json: + print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) + else: + print(f"Database unavailable for risk governor sample: {exc.__class__.__name__}") + return 2 + + context = RiskContext.from_account_state( + build_account_risk_state_from_env(env=env), + market_state=market_state, + backtest_quality_gate=quality_gate, + block_opening_volatility=_get_bool(env, "BLOCK_OPENING_VOLATILITY", False), + ) + decision = RiskGovernorService().evaluate( + context=context, + execution_mode=execution_mode, + new_position=True, + ) + payload = { + "passed": True, + "instrument_id": instrument_id, + "timeframe": timeframe, + "execution_mode": execution_mode, + "decision": decision.to_dict(), + } + if output_json: + print(json.dumps(payload, default=str, ensure_ascii=False, sort_keys=True)) + else: + print(f"Risk governor allowed: {decision.allowed}; reject_codes: {decision.reject_codes}") + return 0 + + +def build_news_windows_from_env(env=None) -> list[dict]: + env = os.environ if env is None else env + if not _get_bool(env, "NEWS_WINDOW_ACTIVE", False): + return [] + now = datetime.now(timezone.utc) + return [ + { + "id": str(env.get("NEWS_WINDOW_ID") or "env_news_window"), + "start_ts": now - timedelta(minutes=_get_int(env, "NEWS_WINDOW_MINUTES_BEFORE", 5)), + "end_ts": now + timedelta(minutes=_get_int(env, "NEWS_WINDOW_MINUTES_AFTER", 5)), + } + ] + + +def _get_int(env, key: str, default: int) -> int: + value = env.get(key) + if value is None or value == "": + return default + return int(value) + + +def _get_bool(env, key: str, default: bool) -> bool: + value = env.get(key) + if value is None or value == "": + return default + return str(value).strip().lower() in {"1", "true", "yes", "on"} + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/repositories/__init__.py b/src/repositories/__init__.py index d62c170..3782933 100644 --- a/src/repositories/__init__.py +++ b/src/repositories/__init__.py @@ -8,6 +8,7 @@ from src.repositories.displacement_repository import DisplacementRepository from src.repositories.execution_repository import ExecutionRepository from src.repositories.fvg_repository import FVGRepository from src.repositories.liquidity_pool_repository import LiquidityPoolRepository +from src.repositories.market_state_repository import MarketStateRepository from src.repositories.market_session_repository import MarketSessionRepository from src.repositories.signal_repository import SignalRepository from src.repositories.structure_repository import StructureRepository @@ -25,6 +26,7 @@ __all__ = [ "ExecutionRepository", "FVGRepository", "LiquidityPoolRepository", + "MarketStateRepository", "MarketSessionRepository", "SignalRepository", "StructureRepository", diff --git a/src/repositories/market_state_repository.py b/src/repositories/market_state_repository.py new file mode 100644 index 0000000..b6c9222 --- /dev/null +++ b/src/repositories/market_state_repository.py @@ -0,0 +1,62 @@ +from typing import Optional + +from sqlalchemy import text + +from src.db.session import SessionLocal + + +class MarketStateRepository: + def fetch_latest_candle(self, instrument_id: int, timeframe: str) -> Optional[dict]: + sql = text( + """ + select id, instrument_id, timeframe, ts_open, ts_close, open, high, low, close, volume, source + from candles + where instrument_id = :instrument_id + and timeframe = :timeframe + order by ts_close desc, id desc + limit 1 + """ + ) + with SessionLocal() as session: + row = session.execute( + sql, + {"instrument_id": instrument_id, "timeframe": timeframe}, + ).mappings().first() + return dict(row) if row else None + + def fetch_session_at(self, instrument_id: int, ts) -> Optional[dict]: + sql = text( + """ + select id, session_code, status, start_ts, end_ts, high, low, midpoint, meta + from market_sessions + where instrument_id = :instrument_id + and start_ts <= :ts + and end_ts > :ts + order by start_ts desc, id desc + limit 1 + """ + ) + with SessionLocal() as session: + row = session.execute(sql, {"instrument_id": instrument_id, "ts": ts}).mappings().first() + return dict(row) if row else None + + def fetch_latest_backtest_quality_gate(self, instrument_id: int) -> Optional[dict]: + sql = text( + """ + select id, summary, ended_ts, started_ts + from backtest_runs + where instrument_id = :instrument_id + order by coalesce(ended_ts, started_ts) desc, id desc + limit 1 + """ + ) + with SessionLocal() as session: + row = session.execute(sql, {"instrument_id": instrument_id}).mappings().first() + if not row: + return None + payload = dict(row) + summary = dict(payload.get("summary") or {}) + quality_gate = dict(summary.get("quality_gate") or {}) + quality_gate["backtest_run_id"] = payload.get("id") + quality_gate["observed_ts"] = payload.get("ended_ts") or payload.get("started_ts") + return quality_gate diff --git a/src/services/execution/__init__.py b/src/services/execution/__init__.py index 453da14..562be5c 100644 --- a/src/services/execution/__init__.py +++ b/src/services/execution/__init__.py @@ -5,6 +5,7 @@ from src.services.execution.execution_service import ( build_execution_observation_from_backtest_result, ExecutionDecision, ExecutionService, + RiskContext, classify_execution_lifecycle, compute_entry_price, evaluate_account_risk_limits, @@ -17,6 +18,7 @@ __all__ = [ "build_execution_observation_from_backtest_result", "ExecutionDecision", "ExecutionService", + "RiskContext", "classify_execution_lifecycle", "compute_entry_price", "evaluate_account_risk_limits", diff --git a/src/services/execution/execution_service.py b/src/services/execution/execution_service.py index e4862f3..f0515f6 100644 --- a/src/services/execution/execution_service.py +++ b/src/services/execution/execution_service.py @@ -12,6 +12,7 @@ from src.services.backtest import ( BACKTEST_LIFECYCLE_TP2_CLOSED, classify_backtest_lifecycle_state_from_result, ) +from src.services.risk import AccountRiskState, RiskContext, RiskGovernorService ALLOWED_EXECUTION_SIGNAL_STATUSES = {"awaiting_confirmation"} @@ -31,19 +32,6 @@ class ExecutionDecision: meta: dict -@dataclass(frozen=True) -class AccountRiskState: - kill_switch: bool = False - daily_pnl_pct: float = 0.0 - consecutive_losses: int = 0 - open_positions: int = 0 - same_side_open_positions: int = 0 - max_daily_loss_pct: float = 0.02 - max_consecutive_losses: int = 3 - max_open_positions: int = 1 - max_same_side_positions: int = 1 - - class ExecutionService: def __init__(self, repository: Optional[ExecutionRepository] = None) -> None: self.repository = repository or ExecutionRepository() @@ -297,6 +285,7 @@ class ExecutionService: account_equity: float = 10000.0, risk_pct: float = 0.01, account_state: Optional[AccountRiskState] = None, + risk_context: Optional[RiskContext] = None, ) -> Optional[ExecutionDecision]: signal = self.repository.fetch_signal_for_execution(signal_id) if not signal: @@ -307,6 +296,7 @@ class ExecutionService: account_equity=account_equity, risk_pct=risk_pct, account_state=account_state, + risk_context=risk_context, ) def create_execution( @@ -317,6 +307,7 @@ class ExecutionService: risk_pct: float = 0.01, confirmed: bool = False, account_state: Optional[AccountRiskState] = None, + risk_context: Optional[RiskContext] = None, ) -> Optional[int]: decision = self.prepare_execution( signal_id=signal_id, @@ -324,6 +315,7 @@ class ExecutionService: account_equity=account_equity, risk_pct=risk_pct, account_state=account_state, + risk_context=risk_context, ) if decision is None or not decision.allowed: return None @@ -376,6 +368,7 @@ def evaluate_execution_decision( account_equity: float, risk_pct: float, account_state: Optional[AccountRiskState] = None, + risk_context: Optional[RiskContext] = None, ) -> ExecutionDecision: reasons: list[str] = [] meta = { @@ -405,8 +398,22 @@ def evaluate_execution_decision( reasons.append("bias_execution_intensity_low") if risk_pct <= 0 or risk_pct > 0.02: reasons.append("risk_pct_out_of_bounds") + if risk_context is None and isinstance(account_state, RiskContext): + risk_context = account_state + if account_state is None and risk_context is not None: + account_state = risk_context + account_risk_reasons = evaluate_account_risk_limits(account_state) reasons.extend(account_risk_reasons) + if risk_context is not None: + risk_decision = RiskGovernorService().evaluate( + context=risk_context, + execution_mode=execution_mode, + new_position=True, + include_account_limits=False, + ) + reasons.extend(risk_decision.reject_codes) + meta["risk_governor"] = risk_decision.to_dict() entry_price = compute_entry_price(signal) stop_loss = float(signal.get("stop_loss") or 0.0) @@ -424,7 +431,7 @@ def evaluate_execution_decision( meta["entry_high"] = float(signal.get("entry_high") or 0.0) meta["risk_distance"] = risk_distance meta["account_equity"] = account_equity - meta["account_risk"] = account_state.__dict__ if account_state else None + meta["account_risk"] = account_state.to_dict() if hasattr(account_state, "to_dict") else (account_state.__dict__ if account_state else None) allowed = len(reasons) == 0 requires_confirmation = execution_mode == "semi_auto" diff --git a/src/services/market_state/__init__.py b/src/services/market_state/__init__.py new file mode 100644 index 0000000..a8a9efc --- /dev/null +++ b/src/services/market_state/__init__.py @@ -0,0 +1,31 @@ +"""Market state service package.""" + +from src.services.market_state.market_state_service import ( + DEFAULT_ALLOWED_EXECUTION_SESSIONS, + DEFAULT_MAX_CANDLE_AGE_SECONDS, + FreshnessState, + MarketState, + MarketStateService, + NewsWindowState, + OpeningVolatilityState, + SessionState, + build_session_state, + evaluate_freshness, + evaluate_news_window, + evaluate_opening_volatility, +) + +__all__ = [ + "DEFAULT_ALLOWED_EXECUTION_SESSIONS", + "DEFAULT_MAX_CANDLE_AGE_SECONDS", + "FreshnessState", + "MarketState", + "MarketStateService", + "NewsWindowState", + "OpeningVolatilityState", + "SessionState", + "build_session_state", + "evaluate_freshness", + "evaluate_news_window", + "evaluate_opening_volatility", +] diff --git a/src/services/market_state/market_state_service.py b/src/services/market_state/market_state_service.py new file mode 100644 index 0000000..4277ae1 --- /dev/null +++ b/src/services/market_state/market_state_service.py @@ -0,0 +1,245 @@ +from dataclasses import asdict, dataclass +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from typing import Optional + +from src.repositories.market_state_repository import MarketStateRepository + + +DEFAULT_MAX_CANDLE_AGE_SECONDS = 300 +DEFAULT_ALLOWED_EXECUTION_SESSIONS = ("LONDON", "NY_AM", "NY_PM") + + +@dataclass(frozen=True) +class FreshnessState: + is_fresh: bool + code: str + latest_ts: Optional[object] + age_seconds: Optional[float] + max_age_seconds: int + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +@dataclass(frozen=True) +class SessionState: + session_code: str + status: str + is_active: bool + allows_new_positions: bool + start_ts: Optional[object] + end_ts: Optional[object] + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +@dataclass(frozen=True) +class NewsWindowState: + is_active: bool + code: str + window_id: Optional[str] + start_ts: Optional[object] + end_ts: Optional[object] + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +@dataclass(frozen=True) +class OpeningVolatilityState: + is_active: bool + code: str + session_code: str + minutes_since_open: Optional[float] + guard_minutes: int + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +@dataclass(frozen=True) +class MarketState: + instrument_id: int + timeframe: str + observed_ts: object + latest_candle: Optional[dict] + freshness: FreshnessState + session: SessionState + news_window: NewsWindowState + opening_volatility: OpeningVolatilityState + + def to_dict(self) -> dict: + return _serialize( + { + "instrument_id": self.instrument_id, + "timeframe": self.timeframe, + "observed_ts": self.observed_ts, + "latest_candle": dict(self.latest_candle or {}), + "freshness": self.freshness.to_dict(), + "session": self.session.to_dict(), + "news_window": self.news_window.to_dict(), + "opening_volatility": self.opening_volatility.to_dict(), + } + ) + + +class MarketStateService: + def __init__(self, repository: Optional[MarketStateRepository] = None) -> None: + self.repository = repository or MarketStateRepository() + + def build_market_state( + self, + *, + instrument_id: int, + timeframe: str = "1m", + now: Optional[datetime] = None, + max_candle_age_seconds: int = DEFAULT_MAX_CANDLE_AGE_SECONDS, + news_windows: Optional[list[dict]] = None, + opening_volatility_minutes: int = 0, + ) -> MarketState: + observed_ts = _ensure_aware(now or datetime.now(timezone.utc)) + latest_candle = self.repository.fetch_latest_candle(instrument_id=instrument_id, timeframe=timeframe) + latest_ts = latest_candle.get("ts_close") if latest_candle else None + freshness = evaluate_freshness( + latest_ts=latest_ts, + now=observed_ts, + max_age_seconds=max_candle_age_seconds, + ) + session_row = self.repository.fetch_session_at(instrument_id=instrument_id, ts=observed_ts) + session = build_session_state(session_row) + news_window = evaluate_news_window(now=observed_ts, windows=news_windows or []) + opening_volatility = evaluate_opening_volatility( + now=observed_ts, + session=session, + opening_volatility_minutes=opening_volatility_minutes, + ) + return MarketState( + instrument_id=instrument_id, + timeframe=timeframe, + observed_ts=observed_ts, + latest_candle=latest_candle, + freshness=freshness, + session=session, + news_window=news_window, + opening_volatility=opening_volatility, + ) + + +def evaluate_freshness(*, latest_ts, now: datetime, max_age_seconds: int) -> FreshnessState: + if latest_ts is None: + return FreshnessState( + is_fresh=False, + code="market_data_missing", + latest_ts=None, + age_seconds=None, + max_age_seconds=max_age_seconds, + ) + normalized_latest = _ensure_aware(latest_ts) + normalized_now = _ensure_aware(now) + age_seconds = max((normalized_now - normalized_latest).total_seconds(), 0.0) + is_fresh = age_seconds <= max_age_seconds + return FreshnessState( + is_fresh=is_fresh, + code="market_data_fresh" if is_fresh else "market_data_stale", + latest_ts=normalized_latest, + age_seconds=age_seconds, + max_age_seconds=max_age_seconds, + ) + + +def build_session_state(session_row: Optional[dict]) -> SessionState: + if not session_row: + return SessionState( + session_code="OFF_HOURS", + status="missing", + is_active=False, + allows_new_positions=False, + start_ts=None, + end_ts=None, + ) + session_code = str(session_row.get("session_code") or "UNKNOWN") + return SessionState( + session_code=session_code, + status=str(session_row.get("status") or "unknown"), + is_active=True, + allows_new_positions=session_code in DEFAULT_ALLOWED_EXECUTION_SESSIONS, + start_ts=session_row.get("start_ts"), + end_ts=session_row.get("end_ts"), + ) + + +def evaluate_news_window(*, now: datetime, windows: list[dict]) -> NewsWindowState: + normalized_now = _ensure_aware(now) + for index, window in enumerate(windows): + start_ts = _ensure_aware(window.get("start_ts")) + end_ts = _ensure_aware(window.get("end_ts")) + if start_ts <= normalized_now <= end_ts: + return NewsWindowState( + is_active=True, + code="news_window_active", + window_id=str(window.get("id") or f"window:{index}"), + start_ts=start_ts, + end_ts=end_ts, + ) + return NewsWindowState( + is_active=False, + code="news_window_inactive", + window_id=None, + start_ts=None, + end_ts=None, + ) + + +def evaluate_opening_volatility( + *, + now: datetime, + session: SessionState, + opening_volatility_minutes: int, +) -> OpeningVolatilityState: + if not session.is_active or not session.start_ts or opening_volatility_minutes <= 0: + return OpeningVolatilityState( + is_active=False, + code="opening_volatility_inactive", + session_code=session.session_code, + minutes_since_open=None, + guard_minutes=opening_volatility_minutes, + ) + minutes_since_open = (_ensure_aware(now) - _ensure_aware(session.start_ts)).total_seconds() / 60 + is_active = 0 <= minutes_since_open <= opening_volatility_minutes + return OpeningVolatilityState( + is_active=is_active, + code="opening_volatility_active" if is_active else "opening_volatility_inactive", + session_code=session.session_code, + minutes_since_open=minutes_since_open, + guard_minutes=opening_volatility_minutes, + ) + + +def _ensure_aware(value) -> datetime: + if isinstance(value, datetime): + parsed = value + elif isinstance(value, str): + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + else: + raise ValueError("datetime_value_required") + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _serialize(value): + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, dict): + return {key: _serialize(item) for key, item in value.items()} + if isinstance(value, list): + return [_serialize(item) for item in value] + if isinstance(value, tuple): + return [_serialize(item) for item in value] + if isinstance(value, Decimal): + return float(value) + if isinstance(value, timedelta): + return value.total_seconds() + return value diff --git a/src/services/risk/__init__.py b/src/services/risk/__init__.py new file mode 100644 index 0000000..72fa388 --- /dev/null +++ b/src/services/risk/__init__.py @@ -0,0 +1,16 @@ +"""Risk governor service package.""" + +from src.services.risk.risk_context import AccountRiskState, RiskContext +from src.services.risk.risk_governor_service import ( + RiskDecision, + RiskGovernorService, + build_risk_rejection, +) + +__all__ = [ + "AccountRiskState", + "RiskContext", + "RiskDecision", + "RiskGovernorService", + "build_risk_rejection", +] diff --git a/src/services/risk/risk_context.py b/src/services/risk/risk_context.py new file mode 100644 index 0000000..10ef10b --- /dev/null +++ b/src/services/risk/risk_context.py @@ -0,0 +1,68 @@ +from dataclasses import asdict, dataclass +from datetime import datetime +from decimal import Decimal +from typing import Optional + + +@dataclass(frozen=True) +class AccountRiskState: + kill_switch: bool = False + daily_pnl_pct: float = 0.0 + consecutive_losses: int = 0 + open_positions: int = 0 + same_side_open_positions: int = 0 + max_daily_loss_pct: float = 0.02 + max_consecutive_losses: int = 3 + max_open_positions: int = 1 + max_same_side_positions: int = 1 + + def to_dict(self) -> dict: + return dict(asdict(self)) + + +@dataclass(frozen=True) +class RiskContext(AccountRiskState): + market_state: Optional[object] = None + backtest_quality_gate: Optional[dict] = None + news_window: Optional[dict] = None + opening_volatility: Optional[dict] = None + block_opening_volatility: bool = False + + @classmethod + def from_account_state( + cls, + account_state: Optional[AccountRiskState] = None, + **kwargs, + ) -> "RiskContext": + source = account_state or AccountRiskState() + return cls( + kill_switch=source.kill_switch, + daily_pnl_pct=source.daily_pnl_pct, + consecutive_losses=source.consecutive_losses, + open_positions=source.open_positions, + same_side_open_positions=source.same_side_open_positions, + max_daily_loss_pct=source.max_daily_loss_pct, + max_consecutive_losses=source.max_consecutive_losses, + max_open_positions=source.max_open_positions, + max_same_side_positions=source.max_same_side_positions, + **kwargs, + ) + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +def _serialize(value): + if hasattr(value, "to_dict"): + return value.to_dict() + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, dict): + return {key: _serialize(item) for key, item in value.items()} + if isinstance(value, list): + return [_serialize(item) for item in value] + if isinstance(value, tuple): + return [_serialize(item) for item in value] + if isinstance(value, Decimal): + return float(value) + return value diff --git a/src/services/risk/risk_governor_service.py b/src/services/risk/risk_governor_service.py new file mode 100644 index 0000000..76f8217 --- /dev/null +++ b/src/services/risk/risk_governor_service.py @@ -0,0 +1,136 @@ +from dataclasses import dataclass +from typing import Optional + +from src.services.risk.risk_context import RiskContext + + +RISK_CODE_EXECUTION_KILL_SWITCH_ENABLED = "execution_kill_switch_enabled" +RISK_CODE_DAILY_LOSS_LIMIT_REACHED = "daily_loss_limit_reached" +RISK_CODE_CONSECUTIVE_LOSS_LIMIT_REACHED = "consecutive_loss_limit_reached" +RISK_CODE_MAX_OPEN_POSITIONS_REACHED = "max_open_positions_reached" +RISK_CODE_SAME_SIDE_EXPOSURE_LIMIT_REACHED = "same_side_exposure_limit_reached" +RISK_CODE_MARKET_DATA_MISSING = "market_data_missing" +RISK_CODE_MARKET_DATA_STALE = "market_data_stale" +RISK_CODE_NEWS_WINDOW_ACTIVE = "news_window_active" +RISK_CODE_OPENING_VOLATILITY_ACTIVE = "opening_volatility_active" +RISK_CODE_BACKTEST_GATE_FAILED_SEMI_AUTO = "backtest_quality_gate_failed_semi_auto" +RISK_CODE_SESSION_NOT_ALLOWED = "session_not_allowed" + + +@dataclass(frozen=True) +class RiskDecision: + allowed: bool + status: str + reject_codes: list[str] + reasons: list[str] + meta: dict + + def to_dict(self) -> dict: + return { + "allowed": self.allowed, + "status": self.status, + "reject_codes": list(self.reject_codes), + "reasons": list(self.reasons), + "meta": dict(self.meta), + } + + +class RiskGovernorService: + def evaluate( + self, + *, + context: RiskContext, + execution_mode: str = "paper", + new_position: bool = True, + include_account_limits: bool = True, + ) -> RiskDecision: + reject_codes: list[str] = [] + meta = { + "execution_mode": execution_mode, + "new_position": new_position, + "account": context.to_dict(), + } + + if include_account_limits: + reject_codes.extend(evaluate_account_risk_codes(context)) + + market_state = _market_state_to_dict(context.market_state) + if market_state: + meta["market_state"] = market_state + freshness = dict(market_state.get("freshness") or {}) + freshness_code = freshness.get("code") + if freshness_code in {RISK_CODE_MARKET_DATA_MISSING, RISK_CODE_MARKET_DATA_STALE}: + reject_codes.append(str(freshness_code)) + session = dict(market_state.get("session") or {}) + if new_position and session and session.get("allows_new_positions") is False: + reject_codes.append(RISK_CODE_SESSION_NOT_ALLOWED) + news_window = dict(market_state.get("news_window") or {}) + if new_position and news_window.get("is_active"): + reject_codes.append(RISK_CODE_NEWS_WINDOW_ACTIVE) + opening_volatility = dict(market_state.get("opening_volatility") or {}) + if new_position and context.block_opening_volatility and opening_volatility.get("is_active"): + reject_codes.append(RISK_CODE_OPENING_VOLATILITY_ACTIVE) + + explicit_news_window = dict(context.news_window or {}) + if new_position and explicit_news_window.get("is_active"): + reject_codes.append(RISK_CODE_NEWS_WINDOW_ACTIVE) + meta["news_window"] = explicit_news_window + + explicit_opening_volatility = dict(context.opening_volatility or {}) + if new_position and context.block_opening_volatility and explicit_opening_volatility.get("is_active"): + reject_codes.append(RISK_CODE_OPENING_VOLATILITY_ACTIVE) + meta["opening_volatility"] = explicit_opening_volatility + + quality_gate = dict(context.backtest_quality_gate or {}) + if execution_mode == "semi_auto" and quality_gate and quality_gate.get("passed") is not True: + reject_codes.append(RISK_CODE_BACKTEST_GATE_FAILED_SEMI_AUTO) + meta["backtest_quality_gate"] = quality_gate + + reject_codes = _dedupe(reject_codes) + return build_risk_rejection(reject_codes=reject_codes, meta=meta) + + +def build_risk_rejection(*, reject_codes: list[str], meta: Optional[dict] = None) -> RiskDecision: + codes = _dedupe(reject_codes) + return RiskDecision( + allowed=len(codes) == 0, + status="allowed" if not codes else "rejected", + reject_codes=codes, + reasons=codes, + meta=dict(meta or {}), + ) + + +def evaluate_account_risk_codes(context: RiskContext) -> list[str]: + reject_codes: list[str] = [] + if context.kill_switch: + reject_codes.append(RISK_CODE_EXECUTION_KILL_SWITCH_ENABLED) + if context.daily_pnl_pct <= -abs(context.max_daily_loss_pct): + reject_codes.append(RISK_CODE_DAILY_LOSS_LIMIT_REACHED) + if context.consecutive_losses >= context.max_consecutive_losses: + reject_codes.append(RISK_CODE_CONSECUTIVE_LOSS_LIMIT_REACHED) + if context.open_positions >= context.max_open_positions: + reject_codes.append(RISK_CODE_MAX_OPEN_POSITIONS_REACHED) + if context.same_side_open_positions >= context.max_same_side_positions: + reject_codes.append(RISK_CODE_SAME_SIDE_EXPOSURE_LIMIT_REACHED) + return reject_codes + + +def _market_state_to_dict(market_state) -> dict: + if market_state is None: + return {} + if hasattr(market_state, "to_dict"): + return market_state.to_dict() + return dict(market_state) + + +def _dedupe(values: list[str]) -> list[str]: + result: list[str] = [] + seen: set[str] = set() + for value in values: + code = str(value or "").strip() + if not code or code in seen: + continue + seen.add(code) + result.append(code) + return result diff --git a/tests/unit/test_market_state_and_risk_governor.py b/tests/unit/test_market_state_and_risk_governor.py new file mode 100644 index 0000000..992d495 --- /dev/null +++ b/tests/unit/test_market_state_and_risk_governor.py @@ -0,0 +1,248 @@ +from datetime import datetime, timedelta, timezone +import os +import unittest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") + +from src.services.execution import evaluate_execution_decision +from src.services.market_state import ( + MarketStateService, + build_session_state, + evaluate_freshness, + evaluate_news_window, + evaluate_opening_volatility, +) +from src.services.risk import AccountRiskState, RiskContext, RiskGovernorService + + +class MarketStateServiceTests(unittest.TestCase): + def test_freshness_accepts_recent_candle(self) -> None: + now = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) + + freshness = evaluate_freshness( + latest_ts=now - timedelta(seconds=60), + now=now, + max_age_seconds=300, + ) + + self.assertTrue(freshness.is_fresh) + self.assertEqual(freshness.code, "market_data_fresh") + self.assertEqual(freshness.age_seconds, 60) + + def test_freshness_rejects_stale_or_missing_candle(self) -> None: + now = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) + + stale = evaluate_freshness( + latest_ts=now - timedelta(minutes=10), + now=now, + max_age_seconds=300, + ) + missing = evaluate_freshness(latest_ts=None, now=now, max_age_seconds=300) + + self.assertFalse(stale.is_fresh) + self.assertEqual(stale.code, "market_data_stale") + self.assertFalse(missing.is_fresh) + self.assertEqual(missing.code, "market_data_missing") + + def test_news_window_placeholder_reports_active_window(self) -> None: + now = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) + + news = evaluate_news_window( + now=now, + windows=[ + { + "id": "cpi", + "start_ts": now - timedelta(minutes=5), + "end_ts": now + timedelta(minutes=5), + } + ], + ) + + self.assertTrue(news.is_active) + self.assertEqual(news.code, "news_window_active") + self.assertEqual(news.window_id, "cpi") + + def test_opening_volatility_placeholder_reports_opening_guard(self) -> None: + now = datetime(2026, 5, 6, 10, 3, tzinfo=timezone.utc) + session = build_session_state( + { + "session_code": "LONDON", + "status": "open", + "start_ts": now - timedelta(minutes=3), + "end_ts": now + timedelta(hours=2), + } + ) + + volatility = evaluate_opening_volatility( + now=now, + session=session, + opening_volatility_minutes=5, + ) + + self.assertTrue(volatility.is_active) + self.assertEqual(volatility.code, "opening_volatility_active") + self.assertEqual(volatility.session_code, "LONDON") + + def test_market_state_service_builds_fact_snapshot(self) -> None: + now = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) + service = MarketStateService(repository=_MarketStateStubRepository(now=now)) + + state = service.build_market_state( + instrument_id=1, + timeframe="1m", + now=now, + max_candle_age_seconds=300, + ) + + self.assertTrue(state.freshness.is_fresh) + self.assertEqual(state.session.session_code, "LONDON") + self.assertTrue(state.session.allows_new_positions) + self.assertFalse(state.news_window.is_active) + + +class RiskGovernorServiceTests(unittest.TestCase): + def test_risk_governor_rejects_stale_market_data(self) -> None: + now = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) + market_state = MarketStateService(repository=_MarketStateStubRepository(now=now, stale=True)).build_market_state( + instrument_id=1, + timeframe="1m", + now=now, + max_candle_age_seconds=300, + ) + + decision = RiskGovernorService().evaluate( + context=RiskContext(market_state=market_state), + execution_mode="paper", + ) + + self.assertFalse(decision.allowed) + self.assertIn("market_data_stale", decision.reject_codes) + + def test_risk_governor_rejects_active_news_window_for_new_position(self) -> None: + now = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) + market_state = MarketStateService(repository=_MarketStateStubRepository(now=now)).build_market_state( + instrument_id=1, + timeframe="1m", + now=now, + news_windows=[ + { + "id": "fomc", + "start_ts": now - timedelta(minutes=10), + "end_ts": now + timedelta(minutes=10), + } + ], + ) + + decision = RiskGovernorService().evaluate( + context=RiskContext(market_state=market_state), + execution_mode="paper", + ) + + self.assertFalse(decision.allowed) + self.assertIn("news_window_active", decision.reject_codes) + + def test_risk_governor_rejects_low_quality_backtest_gate_for_semi_auto_only(self) -> None: + context = RiskContext( + backtest_quality_gate={ + "passed": False, + "reasons": ["profit_factor_below_threshold"], + } + ) + + semi_auto = RiskGovernorService().evaluate(context=context, execution_mode="semi_auto") + paper = RiskGovernorService().evaluate(context=context, execution_mode="paper") + + self.assertFalse(semi_auto.allowed) + self.assertIn("backtest_quality_gate_failed_semi_auto", semi_auto.reject_codes) + self.assertTrue(paper.allowed) + self.assertEqual(paper.reject_codes, []) + + def test_risk_governor_keeps_stable_account_limit_codes(self) -> None: + context = RiskContext.from_account_state( + AccountRiskState( + kill_switch=True, + daily_pnl_pct=-0.02, + consecutive_losses=3, + open_positions=1, + same_side_open_positions=1, + ) + ) + + decision = RiskGovernorService().evaluate(context=context) + + self.assertEqual( + decision.reject_codes, + [ + "execution_kill_switch_enabled", + "daily_loss_limit_reached", + "consecutive_loss_limit_reached", + "max_open_positions_reached", + "same_side_exposure_limit_reached", + ], + ) + + def test_execution_decision_consumes_risk_context(self) -> None: + now = datetime(2026, 5, 6, 10, 0, tzinfo=timezone.utc) + market_state = MarketStateService(repository=_MarketStateStubRepository(now=now, stale=True)).build_market_state( + instrument_id=1, + timeframe="1m", + now=now, + max_candle_age_seconds=300, + ) + + decision = evaluate_execution_decision( + signal=_valid_signal(), + execution_mode="paper", + account_equity=10000.0, + risk_pct=0.01, + risk_context=RiskContext(market_state=market_state), + ) + + self.assertFalse(decision.allowed) + self.assertIn("market_data_stale", decision.reasons) + self.assertIn("risk_governor", decision.meta) + + +class _MarketStateStubRepository: + def __init__(self, now: datetime, stale: bool = False) -> None: + self.now = now + self.stale = stale + + def fetch_latest_candle(self, instrument_id: int, timeframe: str): + age = timedelta(minutes=10) if self.stale else timedelta(seconds=30) + return { + "id": 1, + "instrument_id": instrument_id, + "timeframe": timeframe, + "ts_close": self.now - age, + "close": 100.0, + } + + def fetch_session_at(self, instrument_id: int, ts): + return { + "id": 9, + "session_code": "LONDON", + "status": "open", + "start_ts": self.now - timedelta(minutes=15), + "end_ts": self.now + timedelta(hours=2), + } + + +def _valid_signal() -> dict: + return { + "side": "buy", + "status": "awaiting_confirmation", + "entry_type": "limit", + "entry_low": 100.0, + "entry_high": 101.0, + "stop_loss": 99.0, + "tp1": 102.0, + "tp2": 104.0, + "session_code": "LONDON", + "execution_intensity": "medium", + "meta": {"execution_permission": True}, + } + + +if __name__ == "__main__": + unittest.main()