Phase 2 market state risk governor

This commit is contained in:
Codex
2026-05-06 12:32:50 +08:00
parent e47d197ff4
commit 2e20c60775
13 changed files with 953 additions and 15 deletions
+16
View File
@@ -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
+1 -1
View File
@@ -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"
}
+105
View File
@@ -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())
+2
View File
@@ -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",
@@ -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
+2
View File
@@ -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",
+21 -14
View File
@@ -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"
+31
View File
@@ -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",
]
@@ -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
+16
View File
@@ -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",
]
+68
View File
@@ -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
+136
View File
@@ -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
@@ -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()