Codex
832 lines
39 KiB
Python
832 lines
39 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
import threading
|
|
import time
|
|
from typing import Optional
|
|
|
|
from src.config.settings import settings
|
|
from src.services.alerts import AlertDispatcherService
|
|
from src.services.broker import BROKER_STATE_OK, BrokerReadModelService
|
|
from src.services.decision import DecisionPacketService
|
|
from src.services.market_state import MarketStateService, get_market_truth_supervisor
|
|
from src.services.reconciliation import BrokerExecutionReconciliationService
|
|
from src.services.readiness.shadow_paper_sample_collection_service import ShadowPaperSampleCollectionService
|
|
from src.services.research_council import ResearchCouncilService
|
|
from src.services.runtime.runtime_incident_journal_service import RuntimeIncidentJournalService
|
|
from src.services.runtime.runtime_task_watchdog_service import RuntimeTaskWatchdogService
|
|
from src.services.trade_plan_candidates import SetupWatcherService
|
|
from src.services.trading_plans import TradingPlanService
|
|
from src.services.visual_watch import VisualCaptureRetentionService, VisualReviewService
|
|
|
|
|
|
@dataclass
|
|
class RuntimeTaskState:
|
|
name: str
|
|
interval_seconds: int
|
|
last_started_ts: Optional[str] = None
|
|
last_completed_ts: Optional[str] = None
|
|
last_success_ts: Optional[str] = None
|
|
next_run_ts: Optional[str] = None
|
|
last_status: str = "idle"
|
|
last_error: Optional[str] = None
|
|
run_count: int = 0
|
|
success_count: int = 0
|
|
failure_count: int = 0
|
|
details: dict = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"name": self.name,
|
|
"interval_seconds": self.interval_seconds,
|
|
"last_started_ts": self.last_started_ts,
|
|
"last_completed_ts": self.last_completed_ts,
|
|
"last_success_ts": self.last_success_ts,
|
|
"next_run_ts": self.next_run_ts,
|
|
"last_status": self.last_status,
|
|
"last_error": self.last_error,
|
|
"run_count": self.run_count,
|
|
"success_count": self.success_count,
|
|
"failure_count": self.failure_count,
|
|
"details": dict(self.details or {}),
|
|
}
|
|
|
|
|
|
class TradingRuntimeScheduler:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
market_state_service: Optional[MarketStateService] = None,
|
|
trading_plan_service: Optional[TradingPlanService] = None,
|
|
setup_watcher_service: Optional[SetupWatcherService] = None,
|
|
broker_read_model_service: Optional[BrokerReadModelService] = None,
|
|
reconciliation_service: Optional[BrokerExecutionReconciliationService] = None,
|
|
decision_packet_service: Optional[DecisionPacketService] = None,
|
|
alert_dispatcher_service: Optional[AlertDispatcherService] = None,
|
|
research_council_service: Optional[ResearchCouncilService] = None,
|
|
visual_review_service: Optional[VisualReviewService] = None,
|
|
visual_capture_retention_service: Optional[VisualCaptureRetentionService] = None,
|
|
shadow_paper_sample_service: Optional[ShadowPaperSampleCollectionService] = None,
|
|
runtime_task_watchdog_service: Optional[RuntimeTaskWatchdogService] = None,
|
|
runtime_incident_journal_service: Optional[RuntimeIncidentJournalService] = None,
|
|
) -> None:
|
|
self.market_state_service = market_state_service or MarketStateService(market_truth_supervisor=get_market_truth_supervisor())
|
|
self.trading_plan_service = trading_plan_service or TradingPlanService()
|
|
self.setup_watcher_service = setup_watcher_service or SetupWatcherService()
|
|
self.broker_read_model_service = broker_read_model_service or BrokerReadModelService()
|
|
self.reconciliation_service = reconciliation_service or BrokerExecutionReconciliationService(
|
|
broker_read_model_service=self.broker_read_model_service,
|
|
)
|
|
self.decision_packet_service = decision_packet_service or DecisionPacketService()
|
|
self.alert_dispatcher_service = alert_dispatcher_service or AlertDispatcherService()
|
|
self.research_council_service = research_council_service or ResearchCouncilService()
|
|
self.visual_review_service = visual_review_service or VisualReviewService()
|
|
self.visual_capture_retention_service = visual_capture_retention_service or VisualCaptureRetentionService()
|
|
self.shadow_paper_sample_service = shadow_paper_sample_service or ShadowPaperSampleCollectionService()
|
|
self.runtime_task_watchdog_service = runtime_task_watchdog_service or RuntimeTaskWatchdogService(
|
|
stale_multiplier=settings.runtime_task_stale_multiplier,
|
|
min_stuck_seconds=settings.runtime_task_stuck_min_seconds,
|
|
)
|
|
self.runtime_incident_journal_service = runtime_incident_journal_service or RuntimeIncidentJournalService()
|
|
self.symbols = [item.strip() for item in settings.trading_runtime_symbols.split(",") if item.strip()]
|
|
self.bias_tf = settings.trading_runtime_bias_timeframe
|
|
self.setup_tf = settings.trading_runtime_setup_timeframe
|
|
self.trigger_tf = settings.trading_runtime_trigger_timeframe
|
|
self.tasks = {
|
|
"market_state": RuntimeTaskState("market_state", settings.trading_runtime_market_state_interval_seconds),
|
|
"broker_sync": RuntimeTaskState("broker_sync", settings.trading_runtime_broker_sync_interval_seconds),
|
|
"reconciliation": RuntimeTaskState("reconciliation", settings.trading_runtime_reconciliation_interval_seconds),
|
|
"watch": RuntimeTaskState("watch", settings.trading_runtime_watch_interval_seconds),
|
|
"research_council": RuntimeTaskState("research_council", settings.trading_runtime_research_council_interval_seconds),
|
|
"decision_packet": RuntimeTaskState("decision_packet", settings.trading_runtime_decision_packet_interval_seconds),
|
|
"shadow_paper_observation": RuntimeTaskState("shadow_paper_observation", settings.trading_runtime_decision_packet_interval_seconds),
|
|
"visual_watch": RuntimeTaskState("visual_watch", settings.trading_runtime_decision_packet_interval_seconds),
|
|
"visual_capture_cleanup": RuntimeTaskState("visual_capture_cleanup", settings.visual_capture_cleanup_interval_seconds),
|
|
"alert_dispatch": RuntimeTaskState("alert_dispatch", settings.alert_dispatch_interval_seconds),
|
|
"plan": RuntimeTaskState("plan", settings.trading_runtime_plan_interval_seconds),
|
|
}
|
|
self.current_blockers: list[str] = []
|
|
self.current_warnings: list[str] = []
|
|
self.current_noisy_incidents: list[str] = []
|
|
self.degraded_mode_state: dict = {}
|
|
self.incidents: list[dict] = []
|
|
self.latest_market_state_by_symbol: dict[str, dict] = {}
|
|
self.latest_broker_state_by_symbol: dict[str, dict] = {}
|
|
self.runtime_health: dict = {}
|
|
self.runtime_watchdog: dict = {}
|
|
|
|
def run_cycle(self) -> dict:
|
|
if not settings.trading_runtime_enabled:
|
|
self.current_blockers = ["trading_runtime_disabled"]
|
|
self.current_warnings = []
|
|
self.current_noisy_incidents = []
|
|
snapshot = self.status_snapshot()
|
|
self.runtime_health = dict(snapshot.get("runtime_health") or {})
|
|
self.degraded_mode_state = dict(snapshot.get("degraded_mode_state") or {})
|
|
self.runtime_watchdog = dict(snapshot.get("runtime_watchdog") or {})
|
|
return snapshot
|
|
|
|
blockers: list[str] = []
|
|
warnings: list[str] = []
|
|
noisy_incidents: list[str] = []
|
|
now = datetime.now(timezone.utc)
|
|
for symbol in self.symbols:
|
|
instrument = self.trading_plan_service.repository.fetch_instrument_by_symbol(symbol)
|
|
if not instrument:
|
|
blockers.append(f"missing_instrument:{symbol}")
|
|
continue
|
|
instrument_id = int(instrument["id"])
|
|
self._run_plan_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers)
|
|
market_payload = self._run_market_state_task(
|
|
symbol=symbol,
|
|
instrument_id=instrument_id,
|
|
now=now,
|
|
blockers=blockers,
|
|
warnings=warnings,
|
|
noisy_incidents=noisy_incidents,
|
|
)
|
|
self._run_broker_sync_task(
|
|
symbol=symbol,
|
|
instrument_id=instrument_id,
|
|
now=now,
|
|
blockers=blockers,
|
|
warnings=warnings,
|
|
)
|
|
self._run_reconciliation_task(
|
|
symbol=symbol,
|
|
instrument_id=instrument_id,
|
|
now=now,
|
|
blockers=blockers,
|
|
warnings=warnings,
|
|
)
|
|
self._run_watch_task(
|
|
symbol=symbol,
|
|
instrument_id=instrument_id,
|
|
now=now,
|
|
market_payload=market_payload,
|
|
blockers=blockers,
|
|
warnings=warnings,
|
|
noisy_incidents=noisy_incidents,
|
|
)
|
|
self._run_research_council_task(
|
|
symbol=symbol,
|
|
instrument_id=instrument_id,
|
|
now=now,
|
|
warnings=warnings,
|
|
)
|
|
self._run_decision_packet_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers)
|
|
self._run_shadow_paper_observation_task(
|
|
symbol=symbol,
|
|
instrument_id=instrument_id,
|
|
now=now,
|
|
blockers=blockers,
|
|
warnings=warnings,
|
|
)
|
|
self._run_visual_watch_task(symbol=symbol, instrument_id=instrument_id, now=now, warnings=warnings)
|
|
self._run_visual_capture_cleanup_task(now=now, warnings=warnings)
|
|
self._run_alert_dispatch_task(now=now, blockers=blockers, warnings=warnings)
|
|
|
|
self.current_blockers = _dedupe(blockers)
|
|
self.current_warnings = _dedupe(warnings)
|
|
self.current_noisy_incidents = _dedupe(noisy_incidents)
|
|
snapshot = self.status_snapshot()
|
|
self.runtime_health = dict(snapshot.get("runtime_health") or {})
|
|
self.degraded_mode_state = dict(snapshot.get("degraded_mode_state") or {})
|
|
self.runtime_watchdog = dict(snapshot.get("runtime_watchdog") or {})
|
|
return snapshot
|
|
|
|
def status_snapshot(self) -> dict:
|
|
runtime_watchdog = (
|
|
self.runtime_task_watchdog_service.build_watchdog(self.tasks)
|
|
if settings.runtime_task_watchdog_enabled
|
|
else {"checked_ts": datetime.now(timezone.utc).isoformat(), "stalled_task_count": 0, "stalled_task_names": [], "stalled_tasks": [], "warning_codes": []}
|
|
)
|
|
current_warnings = _dedupe([*self.current_warnings, *list(runtime_watchdog.get("warning_codes") or [])])
|
|
runtime_health = _build_runtime_health(self.current_blockers, current_warnings, self.current_noisy_incidents)
|
|
degraded_mode_state = {
|
|
"is_degraded": bool(self.current_blockers or current_warnings),
|
|
"codes": [*self.current_blockers, *current_warnings],
|
|
}
|
|
snapshot = {
|
|
"enabled": bool(settings.trading_runtime_enabled),
|
|
"scheduler_state": "running" if settings.trading_runtime_enabled else "disabled",
|
|
"symbols": list(self.symbols),
|
|
"timeframe_set": {
|
|
"bias_tf": self.bias_tf,
|
|
"setup_tf": self.setup_tf,
|
|
"trigger_tf": self.trigger_tf,
|
|
},
|
|
"tasks": {name: task.to_dict() for name, task in self.tasks.items()},
|
|
"current_blockers": list(self.current_blockers),
|
|
"current_warnings": list(current_warnings),
|
|
"current_noisy_incidents": list(self.current_noisy_incidents),
|
|
"runtime_health": runtime_health,
|
|
"degraded_mode_state": degraded_mode_state,
|
|
"runtime_watchdog": runtime_watchdog,
|
|
"incidents": list(self.incidents[-50:]),
|
|
"latest_market_state_by_symbol": dict(self.latest_market_state_by_symbol),
|
|
"latest_broker_state_by_symbol": dict(self.latest_broker_state_by_symbol),
|
|
}
|
|
snapshot["incident_summary"] = self.runtime_incident_journal_service.build_from_runtime_status(
|
|
snapshot,
|
|
limit=50,
|
|
persist=False,
|
|
)
|
|
return snapshot
|
|
|
|
def _run_plan_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None:
|
|
task = self.tasks["plan"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
candidate = self.trading_plan_service.build_plan(
|
|
instrument_id=instrument_id,
|
|
bias_tf=self.bias_tf,
|
|
setup_tf=self.setup_tf,
|
|
trigger_tf=self.trigger_tf,
|
|
plan_date=now.date(),
|
|
)
|
|
plan_id = self.trading_plan_service.persist_plan(candidate)
|
|
task.details = {"symbol": symbol, "instrument_id": instrument_id, "plan_id": plan_id, "status": candidate.status}
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"plan_task_failed:{symbol}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", error=str(exc), observed_ts=now)
|
|
|
|
def _run_market_state_task(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
instrument_id: int,
|
|
now: datetime,
|
|
blockers: list[str],
|
|
warnings: list[str],
|
|
noisy_incidents: list[str],
|
|
) -> dict:
|
|
task = self.tasks["market_state"]
|
|
if not _task_due(task, now):
|
|
return self.latest_market_state_by_symbol.get(symbol, {})
|
|
_task_start(task, now)
|
|
try:
|
|
market_state = self.market_state_service.build_market_state(
|
|
instrument_id=instrument_id,
|
|
timeframe=self.trigger_tf,
|
|
)
|
|
snapshot_id = self.market_state_service.persist_market_state_snapshot(market_state=market_state)
|
|
payload = market_state.to_dict()
|
|
payload["snapshot_id"] = snapshot_id
|
|
self.latest_market_state_by_symbol[symbol] = payload
|
|
task.details = {"symbol": symbol, "instrument_id": instrument_id, "snapshot_id": snapshot_id}
|
|
_task_success(task, now)
|
|
if payload.get("freshness", {}).get("code") != "market_data_fresh":
|
|
code = f"market_state_degraded:{symbol}:{payload.get('freshness', {}).get('code')}"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="warning", observed_ts=now)
|
|
if payload.get("stream_state", {}).get("code") != "market_stream_connected":
|
|
code = f"market_stream_degraded:{symbol}:{payload.get('stream_state', {}).get('code')}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", observed_ts=now)
|
|
if payload.get("data_gap_state", {}).get("code") != "market_data_gap_clear":
|
|
code = f"market_gap_degraded:{symbol}:{payload.get('data_gap_state', {}).get('code')}"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="warning", observed_ts=now)
|
|
return payload
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"market_state_task_failed:{symbol}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", error=str(exc), observed_ts=now)
|
|
return self.latest_market_state_by_symbol.get(symbol, {})
|
|
|
|
def _run_broker_sync_task(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
instrument_id: int,
|
|
now: datetime,
|
|
blockers: list[str],
|
|
warnings: list[str],
|
|
) -> None:
|
|
task = self.tasks["broker_sync"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
snapshot = self.broker_read_model_service.fetch_snapshot(
|
|
instrument_id=instrument_id,
|
|
known_execution_refs={"broker_order_ids": [], "open_execution_count": 0},
|
|
)
|
|
self.latest_broker_state_by_symbol[symbol] = snapshot.to_dict()
|
|
task.details = {"symbol": symbol, "instrument_id": instrument_id, "status": snapshot.status}
|
|
if snapshot.status != BROKER_STATE_OK:
|
|
code = f"broker_sync_degraded:{symbol}:{snapshot.error or snapshot.status}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", error=snapshot.error, observed_ts=now)
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"broker_sync_task_failed:{symbol}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", error=str(exc), observed_ts=now)
|
|
|
|
def _run_watch_task(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
instrument_id: int,
|
|
now: datetime,
|
|
market_payload: dict,
|
|
blockers: list[str],
|
|
warnings: list[str],
|
|
noisy_incidents: list[str],
|
|
) -> None:
|
|
task = self.tasks["watch"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
freshness_code = ((market_payload or {}).get("freshness") or {}).get("code")
|
|
stream_code = ((market_payload or {}).get("stream_state") or {}).get("code")
|
|
gap_code = ((market_payload or {}).get("data_gap_state") or {}).get("code")
|
|
readonly_market_usable = freshness_code == "market_data_fresh" and gap_code == "market_data_gap_clear"
|
|
stream_fallback_active = readonly_market_usable and stream_code != "market_stream_connected"
|
|
if not readonly_market_usable:
|
|
task.details = {"symbol": symbol, "status": "skipped_market_degraded"}
|
|
_task_success(task, now)
|
|
code = f"watch_skipped_market_degraded:{symbol}"
|
|
noisy_incidents.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="noisy_incident", observed_ts=now)
|
|
return
|
|
if stream_fallback_active:
|
|
warnings.append(f"watch_using_rest_fallback:{symbol}:{stream_code}")
|
|
try:
|
|
candidate = self.setup_watcher_service.build_candidate(
|
|
instrument_id=instrument_id,
|
|
setup_tf=self.setup_tf,
|
|
trigger_tf=self.trigger_tf,
|
|
plan_date=now.date(),
|
|
)
|
|
candidate_id = self.setup_watcher_service.persist_candidate(candidate) if candidate is not None else None
|
|
task.details = {
|
|
"symbol": symbol,
|
|
"instrument_id": instrument_id,
|
|
"candidate_id": candidate_id,
|
|
"candidate_generated": bool(candidate),
|
|
"market_input_mode": "rest_fallback" if stream_fallback_active else "stream",
|
|
}
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"watch_task_failed:{symbol}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", error=str(exc), observed_ts=now)
|
|
|
|
def _run_reconciliation_task(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
instrument_id: int,
|
|
now: datetime,
|
|
blockers: list[str],
|
|
warnings: list[str],
|
|
) -> None:
|
|
task = self.tasks["reconciliation"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
summary = self.reconciliation_service.reconcile_instrument(instrument_id=instrument_id, now=now)
|
|
task.details = {
|
|
"symbol": symbol,
|
|
"instrument_id": instrument_id,
|
|
"status": summary.get("status"),
|
|
"matched_execution_ids": list(summary.get("matched_execution_ids") or []),
|
|
"imported_execution_ids": list(summary.get("imported_execution_ids") or []),
|
|
"journal_updates": list(summary.get("journal_updates") or []),
|
|
}
|
|
if summary.get("status") != "ok":
|
|
code = f"reconciliation_degraded:{symbol}:{summary.get('error') or summary.get('status')}"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="warning", error=str(summary.get("error")), observed_ts=now)
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"reconciliation_task_failed:{symbol}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", error=str(exc), observed_ts=now)
|
|
|
|
def _run_decision_packet_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None:
|
|
task = self.tasks["decision_packet"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
packet = self.decision_packet_service.build_persist_and_alert(instrument_id=instrument_id)
|
|
visual_trigger_types: list[str] = []
|
|
visual_trigger_policy_statuses: list[str] = []
|
|
visual_trigger_skip_reasons: list[str] = []
|
|
visual_trigger_policy_reasons: list[str] = []
|
|
visual_triggered_alert_count = 0
|
|
visual_trigger_skipped_alert_count = 0
|
|
for item in list(packet.get("alerts") or []):
|
|
payload = dict(item.get("payload") or {})
|
|
visual_trigger_types.extend(str(value) for value in list(payload.get("visual_trigger_types") or []) if value)
|
|
if payload.get("visual_trigger_type"):
|
|
visual_trigger_types.append(str(payload.get("visual_trigger_type")))
|
|
policy = dict(payload.get("visual_trigger_policy") or {})
|
|
if policy.get("status"):
|
|
policy_status = str(policy.get("status"))
|
|
visual_trigger_policy_statuses.append(policy_status)
|
|
if policy_status == "triggered":
|
|
visual_triggered_alert_count += 1
|
|
if policy_status == "skipped":
|
|
visual_trigger_skipped_alert_count += 1
|
|
if policy.get("skip_reason"):
|
|
visual_trigger_skip_reasons.append(str(policy.get("skip_reason")))
|
|
if policy.get("reason"):
|
|
visual_trigger_policy_reasons.append(str(policy.get("reason")))
|
|
visual_trigger_types = _dedupe(visual_trigger_types)
|
|
visual_trigger_policy_statuses = _dedupe(visual_trigger_policy_statuses)
|
|
visual_trigger_skip_reasons = _dedupe(visual_trigger_skip_reasons)
|
|
visual_trigger_policy_reasons = _dedupe(visual_trigger_policy_reasons)
|
|
task.details = {
|
|
"symbol": symbol,
|
|
"instrument_id": instrument_id,
|
|
"packet_id": packet.get("id"),
|
|
"status": packet.get("status"),
|
|
"action_recommendation": packet.get("action_recommendation"),
|
|
"alerts": len(packet.get("alerts") or []),
|
|
"visual_trigger_types": visual_trigger_types,
|
|
"visual_watch_reserved": bool(visual_trigger_types),
|
|
"visual_trigger_policy_statuses": visual_trigger_policy_statuses,
|
|
"visual_trigger_skip_reasons": visual_trigger_skip_reasons,
|
|
"visual_trigger_policy_reasons": visual_trigger_policy_reasons,
|
|
"visual_triggered_alert_count": visual_triggered_alert_count,
|
|
"visual_trigger_skipped_alert_count": visual_trigger_skipped_alert_count,
|
|
"research_council_present": bool((packet.get("research_council") or {}).get("present")),
|
|
}
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"decision_packet_task_failed:{symbol}"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="blocker", error=str(exc), observed_ts=now)
|
|
|
|
def _run_shadow_paper_observation_task(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
instrument_id: int,
|
|
now: datetime,
|
|
blockers: list[str],
|
|
warnings: list[str],
|
|
) -> None:
|
|
task = self.tasks["shadow_paper_observation"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
payload = self.shadow_paper_sample_service.record_daily_observation(
|
|
instrument_id=instrument_id,
|
|
timeframe=self.trigger_tf,
|
|
created_by="runtime_shadow_collector",
|
|
runtime_status={
|
|
"current_blockers": list(blockers),
|
|
"latest_market_state_by_symbol": dict(self.latest_market_state_by_symbol),
|
|
},
|
|
now=now,
|
|
)
|
|
task.details = {
|
|
"symbol": symbol,
|
|
"instrument_id": instrument_id,
|
|
"status": payload.get("status"),
|
|
"observation_id": payload.get("observation_id"),
|
|
"observation_date": payload.get("observation_date"),
|
|
"observation_status": payload.get("observation_status"),
|
|
"system_recommendation": payload.get("system_recommendation"),
|
|
"operator_agreement": payload.get("operator_agreement"),
|
|
"next_collection_actions": list(payload.get("next_collection_actions") or []),
|
|
}
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"shadow_paper_observation_failed:{symbol}"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="warning", error=str(exc), observed_ts=now)
|
|
|
|
def _run_research_council_task(
|
|
self,
|
|
*,
|
|
symbol: str,
|
|
instrument_id: int,
|
|
now: datetime,
|
|
warnings: list[str],
|
|
) -> None:
|
|
task = self.tasks["research_council"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
payload = self.research_council_service.build_and_persist(
|
|
instrument_id=instrument_id,
|
|
timeframe=self.trigger_tf,
|
|
now=now,
|
|
)
|
|
decision = dict((payload.get("payload") or {}).get("decision") or payload.get("decision") or {})
|
|
task.details = {
|
|
"symbol": symbol,
|
|
"instrument_id": instrument_id,
|
|
"run_id": payload.get("id"),
|
|
"status": payload.get("status") or "completed",
|
|
"overall_stance": decision.get("overall_stance"),
|
|
"committee_confidence": decision.get("committee_confidence"),
|
|
"recommended_watch_items": list(decision.get("recommended_watch_items") or []),
|
|
}
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"research_council_task_failed:{symbol}"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="warning", error=str(exc), observed_ts=now)
|
|
|
|
def _run_visual_watch_task(self, *, symbol: str, instrument_id: int, now: datetime, warnings: list[str]) -> None:
|
|
task = self.tasks["visual_watch"]
|
|
if not _task_due(task, now):
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
payload = self.visual_review_service.review_pending_watch_runs(limit=1, instrument_id=instrument_id)
|
|
items = list(payload.get("items") or [])
|
|
latest = items[-1] if items else {}
|
|
task.details = {
|
|
"symbol": symbol,
|
|
"instrument_id": instrument_id,
|
|
"status": payload.get("status"),
|
|
"skip_reason": payload.get("skip_reason"),
|
|
"attempted": int(payload.get("attempted") or 0),
|
|
"visual_reviews_created": int(payload.get("reviewed") or 0),
|
|
"trigger_type_counts": dict(payload.get("trigger_type_counts") or {}),
|
|
"review_state_counts": dict(payload.get("review_state_counts") or {}),
|
|
"last_visual_state": latest.get("visual_state"),
|
|
"last_snapshot_source": (
|
|
latest.get("snapshot_source_name") or "snapshot_repository"
|
|
if latest.get("snapshot_id")
|
|
else "none"
|
|
),
|
|
"last_snapshot_image_path": latest.get("snapshot_image_path"),
|
|
"last_snapshot_artifact_path": latest.get("snapshot_artifact_path"),
|
|
"last_watch_run_id": latest.get("watch_run_id"),
|
|
}
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = f"visual_watch_task_failed:{symbol}"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol=symbol, code=code, severity="warning", error=str(exc), observed_ts=now)
|
|
|
|
def _run_alert_dispatch_task(self, *, now: datetime, blockers: list[str], warnings: list[str]) -> None:
|
|
task = self.tasks["alert_dispatch"]
|
|
if not _task_due(task, now):
|
|
return
|
|
if not settings.alert_dispatch_enabled:
|
|
task.details = {"status": "disabled"}
|
|
_task_success(task, now)
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
payload = self.alert_dispatcher_service.dispatch_pending(limit=settings.alert_dispatch_max_per_run)
|
|
items = list(payload.get("items") or [])
|
|
failed = [item for item in items if item.get("status") == "failed"]
|
|
silent_window = dict(payload.get("silent_window") or {})
|
|
task.details = {
|
|
"attempted": int(payload.get("attempted") or 0),
|
|
"queued_candidates": int(payload.get("queued_candidates") or 0),
|
|
"retry_candidates": int(payload.get("retry_candidates") or 0),
|
|
"retried": int(payload.get("retried") or 0),
|
|
"deferred_retry_backoff": int(payload.get("deferred_retry_backoff") or 0),
|
|
"retry_exhausted": int(payload.get("retry_exhausted") or 0),
|
|
"transport_retried": int(payload.get("transport_retried") or 0),
|
|
"transport_retry_failed": int(payload.get("transport_retry_failed") or 0),
|
|
"policy_external_allowed": int(payload.get("policy_external_allowed") or 0),
|
|
"workbench_only": int(payload.get("workbench_only") or 0),
|
|
"suppressed_noise_policy": int(payload.get("suppressed_noise_policy") or 0),
|
|
"media_attached": int(payload.get("media_attached") or 0),
|
|
"media_linked": int(payload.get("media_linked") or 0),
|
|
"media_unavailable": int(payload.get("media_unavailable") or 0),
|
|
"channel_disabled": int(payload.get("channel_disabled") or 0),
|
|
"channel_summary": dict(payload.get("channel_summary") or {}),
|
|
"channel_health": dict(payload.get("channel_health") or {}),
|
|
"delivered": sum(1 for item in items if item.get("status") == "delivered"),
|
|
"skipped": sum(1 for item in items if item.get("status") == "skipped"),
|
|
"failed": len(failed),
|
|
"silent_window_active": bool(silent_window.get("active")),
|
|
"silent_window_label": silent_window.get("window_label"),
|
|
"silent_window_next_allowed_ts": silent_window.get("next_allowed_ts"),
|
|
"suppressed_silent_window": int(payload.get("suppressed_silent_window") or 0),
|
|
}
|
|
if failed:
|
|
code = "alert_dispatch_failed"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol="runtime", code=code, severity="warning", error=str(failed[0].get("error") or "dispatch_failed"), observed_ts=now)
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = "alert_dispatch_task_failed"
|
|
blockers.append(code)
|
|
self._record_incident(task_name=task.name, symbol="runtime", code=code, severity="blocker", error=str(exc), observed_ts=now)
|
|
|
|
def _run_visual_capture_cleanup_task(self, *, now: datetime, warnings: list[str]) -> None:
|
|
task = self.tasks["visual_capture_cleanup"]
|
|
if not _task_due(task, now):
|
|
return
|
|
if not settings.visual_capture_cleanup_enabled:
|
|
task.details = {"status": "disabled"}
|
|
_task_success(task, now)
|
|
return
|
|
_task_start(task, now)
|
|
try:
|
|
payload = self.visual_capture_retention_service.cleanup_captures(
|
|
retention_hours=settings.visual_capture_retention_hours,
|
|
max_delete_files=settings.visual_capture_cleanup_max_delete_files,
|
|
dry_run=False,
|
|
)
|
|
task.details = {
|
|
"status": payload.get("status"),
|
|
"root": payload.get("root"),
|
|
"retention_hours": payload.get("retention_hours"),
|
|
"eligible_file_count": int(payload.get("eligible_file_count") or 0),
|
|
"deleted_file_count": int(payload.get("deleted_file_count") or 0),
|
|
"deleted_bytes": int(payload.get("deleted_bytes") or 0),
|
|
"retained_file_count": int(payload.get("retained_file_count") or 0),
|
|
"reason": payload.get("reason"),
|
|
}
|
|
_task_success(task, now)
|
|
except Exception as exc:
|
|
_task_failure(task, now, str(exc))
|
|
code = "visual_capture_cleanup_task_failed"
|
|
warnings.append(code)
|
|
self._record_incident(task_name=task.name, symbol="runtime", code=code, severity="warning", error=str(exc), observed_ts=now)
|
|
|
|
def _record_incident(
|
|
self,
|
|
*,
|
|
task_name: str,
|
|
symbol: str,
|
|
code: str,
|
|
severity: str,
|
|
observed_ts: datetime,
|
|
error: Optional[str] = None,
|
|
) -> None:
|
|
self.incidents.append(
|
|
{
|
|
"task_name": task_name,
|
|
"symbol": symbol,
|
|
"code": code,
|
|
"severity": severity,
|
|
"error": error,
|
|
"observed_ts": observed_ts.isoformat(),
|
|
}
|
|
)
|
|
if len(self.incidents) > 200:
|
|
self.incidents = self.incidents[-200:]
|
|
|
|
|
|
class TradingRuntimeSupervisor:
|
|
def __init__(self, scheduler: Optional[TradingRuntimeScheduler] = None) -> None:
|
|
self.scheduler = scheduler or TradingRuntimeScheduler()
|
|
self._thread: Optional[threading.Thread] = None
|
|
self._stop_event = threading.Event()
|
|
self._lock = threading.RLock()
|
|
self._last_status: dict = self.scheduler.status_snapshot()
|
|
self._started_ts: Optional[str] = None
|
|
|
|
def start(self) -> None:
|
|
if not settings.trading_runtime_enabled:
|
|
return
|
|
with self._lock:
|
|
if self._thread and self._thread.is_alive():
|
|
return
|
|
self._stop_event.clear()
|
|
self._started_ts = datetime.now(timezone.utc).isoformat()
|
|
self._thread = threading.Thread(target=self._run_loop, name="trading-runtime-supervisor", daemon=True)
|
|
self._thread.start()
|
|
|
|
def stop(self) -> None:
|
|
self._stop_event.set()
|
|
with self._lock:
|
|
if self._thread and self._thread.is_alive():
|
|
self._thread.join(timeout=2)
|
|
|
|
def status(self) -> dict:
|
|
with self._lock:
|
|
status = self.scheduler.status_snapshot()
|
|
self._last_status = status
|
|
status["supervisor_state"] = "running" if self._thread and self._thread.is_alive() else "stopped"
|
|
status["started_ts"] = self._started_ts
|
|
return status
|
|
|
|
def run_once(self) -> dict:
|
|
status = self.scheduler.run_cycle()
|
|
with self._lock:
|
|
self._last_status = status
|
|
current = dict(status)
|
|
current["supervisor_state"] = "running" if self._thread and self._thread.is_alive() else "stopped"
|
|
current["started_ts"] = self._started_ts
|
|
return current
|
|
|
|
def _run_loop(self) -> None:
|
|
while not self._stop_event.is_set():
|
|
self.run_once()
|
|
for _ in range(5):
|
|
if self._stop_event.is_set():
|
|
return
|
|
time.sleep(1)
|
|
|
|
|
|
_runtime_supervisor: Optional[TradingRuntimeSupervisor] = None
|
|
|
|
|
|
def get_trading_runtime_service() -> TradingRuntimeSupervisor:
|
|
global _runtime_supervisor
|
|
if _runtime_supervisor is None:
|
|
_runtime_supervisor = TradingRuntimeSupervisor()
|
|
return _runtime_supervisor
|
|
|
|
|
|
def start_trading_runtime_service() -> TradingRuntimeSupervisor:
|
|
service = get_trading_runtime_service()
|
|
service.start()
|
|
return service
|
|
|
|
|
|
def _task_due(task: RuntimeTaskState, now: datetime) -> bool:
|
|
if task.last_started_ts is None:
|
|
return True
|
|
last_started = _parse_iso(task.last_started_ts)
|
|
return (now - last_started).total_seconds() >= max(task.interval_seconds, 1)
|
|
|
|
|
|
def _task_start(task: RuntimeTaskState, now: datetime) -> None:
|
|
task.last_started_ts = now.isoformat()
|
|
task.last_status = "running"
|
|
task.last_error = None
|
|
task.run_count += 1
|
|
|
|
|
|
def _task_success(task: RuntimeTaskState, now: datetime) -> None:
|
|
iso = now.isoformat()
|
|
task.last_completed_ts = iso
|
|
task.last_success_ts = iso
|
|
task.next_run_ts = _next_run_iso(task, now)
|
|
task.last_status = "ok"
|
|
task.last_error = None
|
|
task.success_count += 1
|
|
|
|
|
|
def _task_failure(task: RuntimeTaskState, now: datetime, error: str) -> None:
|
|
task.last_completed_ts = now.isoformat()
|
|
task.next_run_ts = _next_run_iso(task, now)
|
|
task.last_status = "error"
|
|
task.last_error = error
|
|
task.failure_count += 1
|
|
|
|
|
|
def _parse_iso(value: str) -> datetime:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
|
|
|
|
|
|
def _next_run_iso(task: RuntimeTaskState, now: datetime) -> str:
|
|
return datetime.fromtimestamp(now.timestamp() + max(task.interval_seconds, 1), tz=timezone.utc).isoformat()
|
|
|
|
|
|
def _dedupe(values: list[str]) -> list[str]:
|
|
result = []
|
|
seen = set()
|
|
for item in values:
|
|
text = str(item or "").strip()
|
|
if not text or text in seen:
|
|
continue
|
|
seen.add(text)
|
|
result.append(text)
|
|
return result
|
|
|
|
|
|
def _build_runtime_health(blockers: list[str], warnings: list[str], noisy_incidents: list[str]) -> dict:
|
|
if blockers:
|
|
level = "blocked"
|
|
elif warnings:
|
|
level = "warning"
|
|
elif noisy_incidents:
|
|
level = "noisy"
|
|
else:
|
|
level = "ok"
|
|
return {
|
|
"level": level,
|
|
"blocker_count": len(blockers),
|
|
"warning_count": len(warnings),
|
|
"noisy_incident_count": len(noisy_incidents),
|
|
}
|