alphax/app/core/strategy_contract.py

"""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, is_strategy_allowed_for_side, 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.direction = "short" if str(self.direction or "").strip().lower() == "short" else "long"
        if not is_strategy_allowed_for_side(self.strategy_code, self.direction):
            raise ValueError(f"strategy {self.strategy_code} cannot emit {self.direction} signal")
        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 {},
    }