141 lines
5.4 KiB
Python
141 lines
5.4 KiB
Python
"""Stable strategy signal contract shared by scanners, recommendations and ledgers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from app.config.config_loader import get_meta
|
|
from app.core.factor_roles import factor_roles_for_codes, validate_factor_roles
|
|
from app.core.strategy_registry import MAIN_COMPOSITE_STRATEGY, normalize_strategy_code, strategy_label
|
|
|
|
|
|
def _safe_dict(value: Any) -> dict:
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _safe_float(value: Any, default: float = 0.0) -> float:
|
|
try:
|
|
if value is None or value == "":
|
|
return default
|
|
return float(value)
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
@dataclass
|
|
class StrategySignal:
|
|
strategy_code: str
|
|
symbol: str
|
|
strategy_version: str = ""
|
|
direction: str = "long"
|
|
status: str = "candidate"
|
|
confidence: float = 0.0
|
|
score: float = 0.0
|
|
run_id: str = ""
|
|
trigger: dict[str, Any] = field(default_factory=dict)
|
|
factor_roles: dict[str, str] = field(default_factory=dict)
|
|
entry_plan: dict[str, Any] = field(default_factory=dict)
|
|
risk_plan: dict[str, Any] = field(default_factory=dict)
|
|
decision_log: dict[str, Any] = field(default_factory=dict)
|
|
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
|
|
|
|
def __post_init__(self):
|
|
self.strategy_code = normalize_strategy_code(self.strategy_code)
|
|
self.factor_roles = validate_factor_roles(self.factor_roles)
|
|
|
|
def to_json_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"strategy_code": self.strategy_code,
|
|
"strategy_name": strategy_label(self.strategy_code),
|
|
"strategy_version": self.strategy_version or "",
|
|
"symbol": self.symbol,
|
|
"direction": self.direction or "long",
|
|
"status": self.status or "candidate",
|
|
"confidence": round(_safe_float(self.confidence), 6),
|
|
"score": round(_safe_float(self.score), 6),
|
|
"run_id": self.run_id or "",
|
|
"trigger": _safe_dict(self.trigger),
|
|
"factor_roles": validate_factor_roles(self.factor_roles),
|
|
"entry_plan": _safe_dict(self.entry_plan),
|
|
"risk_plan": _safe_dict(self.risk_plan),
|
|
"decision_log": _safe_dict(self.decision_log),
|
|
"created_at": self.created_at,
|
|
}
|
|
|
|
|
|
def current_strategy_version() -> str:
|
|
try:
|
|
return str(get_meta().get("strategy_version") or "").strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def default_main_composite_signal(
|
|
*,
|
|
symbol: str,
|
|
score: float = 0,
|
|
signal_codes: list[str] | None = None,
|
|
entry_plan: dict | None = None,
|
|
market_context: dict | None = None,
|
|
decision_log: dict | None = None,
|
|
) -> StrategySignal:
|
|
return StrategySignal(
|
|
strategy_code=MAIN_COMPOSITE_STRATEGY,
|
|
strategy_version=current_strategy_version(),
|
|
symbol=symbol,
|
|
status="candidate",
|
|
confidence=max(0.0, min(100.0, _safe_float(score))),
|
|
score=_safe_float(score),
|
|
trigger={
|
|
"source": MAIN_COMPOSITE_STRATEGY,
|
|
"signal_codes": signal_codes or [],
|
|
"market_context": _safe_dict(market_context),
|
|
},
|
|
factor_roles=factor_roles_for_codes(signal_codes or []),
|
|
entry_plan=_safe_dict(entry_plan),
|
|
risk_plan={"source": "entry_quality_and_global_risk"},
|
|
decision_log=_safe_dict(decision_log),
|
|
)
|
|
|
|
|
|
def strategy_context_payload(signal: StrategySignal | dict | None, *, fallback_symbol: str = "", fallback_score: float = 0, signal_codes: list[str] | None = None, entry_plan: dict | None = None, market_context: dict | None = None) -> dict[str, Any]:
|
|
if isinstance(signal, StrategySignal):
|
|
payload = signal.to_json_dict()
|
|
elif isinstance(signal, dict):
|
|
payload = dict(signal)
|
|
else:
|
|
payload = default_main_composite_signal(
|
|
symbol=fallback_symbol,
|
|
score=fallback_score,
|
|
signal_codes=signal_codes or [],
|
|
entry_plan=entry_plan,
|
|
market_context=market_context,
|
|
).to_json_dict()
|
|
|
|
payload["strategy_code"] = normalize_strategy_code(payload.get("strategy_code"))
|
|
payload.setdefault("strategy_name", strategy_label(payload["strategy_code"]))
|
|
payload.setdefault("strategy_version", current_strategy_version())
|
|
payload.setdefault("symbol", fallback_symbol or payload.get("symbol") or "")
|
|
payload.setdefault("score", fallback_score)
|
|
payload.setdefault("confidence", payload.get("score") or fallback_score)
|
|
payload["factor_roles"] = validate_factor_roles(payload.get("factor_roles") or factor_roles_for_codes(signal_codes or []))
|
|
payload.setdefault("entry_plan", _safe_dict(entry_plan))
|
|
payload.setdefault("trigger", {})
|
|
payload.setdefault("risk_plan", {})
|
|
payload.setdefault("decision_log", {})
|
|
payload.setdefault("created_at", datetime.now().isoformat())
|
|
return payload
|
|
|
|
|
|
def signal_to_recommendation_context(signal: StrategySignal | dict | None, **fallback) -> dict[str, Any]:
|
|
payload = strategy_context_payload(signal, **fallback)
|
|
return {
|
|
"strategy_code": payload["strategy_code"],
|
|
"strategy_version": payload.get("strategy_version") or current_strategy_version(),
|
|
"strategy_signal_id": int(payload.get("strategy_signal_id") or payload.get("id") or 0),
|
|
"strategy_snapshot": payload,
|
|
"factor_roles": payload.get("factor_roles") or {},
|
|
}
|