strategy_runtime.py

from __future__ import annotations

import os
import math
from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Mapping

from quant_platform_kit import PortfolioSnapshot, Position, build_strategy_evaluation_inputs
from quant_platform_kit.common.strategy_contracts import (
    StrategyContext,
    StrategyDecision,
    StrategyEntrypoint,
    StrategyRuntimeAdapter,
    build_strategy_context_from_available_inputs,
    resolve_strategy_artifact_contract,
)
from quant_platform_kit.risk.contracts import CandidateRiskIdentity

from crypto_strategies import get_platform_runtime_adapter
from live_risk_authority import (
    LiveRiskAuthorityError,
    bind_live_risk_authority,
    config_sha256,
)
from strategy_loader import (
    load_research_strategy_entrypoint_for_profile,
    load_strategy_entrypoint_for_profile,
)
from strategy_registry import BINANCE_PLATFORM, resolve_research_strategy_metadata
from trend_pool_support import get_default_live_pool_candidates as tp_get_default_live_pool_candidates


DEFAULT_LOCAL_TREND_POOL_ARTIFACT = Path(__file__).resolve().parent / "artifacts" / "live_pool_legacy.json"
# Ensure artifacts directory exists so local-fallback path never fails with FileNotFoundError
DEFAULT_LOCAL_TREND_POOL_ARTIFACT.parent.mkdir(parents=True, exist_ok=True)
DEFAULT_TREND_POOL_SIZE = 5
COMBO_RUNTIME_ENV_OVERRIDES: tuple[tuple[str, str, str], ...] = (
    ("BTC_WEIGHT", "btc_weight", "ratio"),
    ("TREND_WEIGHT", "trend_weight", "ratio"),
    ("DYNAMIC_MODE", "dynamic_mode", "bool"),
    ("DYNAMIC_REGIME_MODE", "dynamic_regime_mode", "regime_mode"),
    ("DYNAMIC_REGIME_OFF_CUT", "dynamic_regime_off_cut", "ratio"),
    ("DYNAMIC_HARD_SMA200_RATIO", "dynamic_hard_sma200_ratio", "positive_float"),
    ("DYNAMIC_HARD_MA200_SLOPE", "dynamic_hard_ma200_slope", "float"),
    ("DYNAMIC_SOFT_SMA200_RATIO", "dynamic_soft_sma200_ratio", "positive_float"),
    ("DYNAMIC_HARD_BTC_WEIGHT", "dynamic_hard_btc_weight", "ratio"),
    ("DYNAMIC_HARD_TREND_WEIGHT", "dynamic_hard_trend_weight", "ratio"),
    ("DYNAMIC_SOFT_BTC_WEIGHT", "dynamic_soft_btc_weight", "ratio"),
    ("DYNAMIC_SOFT_TREND_WEIGHT", "dynamic_soft_trend_weight", "ratio"),
    ("ROTATION_TOP_N", "rotation_top_n", "int"),
    ("TARGET_VOL", "target_vol", "positive_float"),
    ("CIRCUIT_BREAKER_ENABLED", "circuit_breaker_enabled", "bool"),
    ("ZSCORE_EXIT_RISK_REDUCED_EXPOSURE", "zscore_exit_risk_reduced_exposure", "ratio"),
    ("ZSCORE_EXIT_RISK_OFF_EXPOSURE", "zscore_exit_risk_off_exposure", "ratio"),
    ("ZSCORE_EXIT_ALLOW_OUTSIDE_EXECUTION_WINDOW", "zscore_exit_allow_outside_execution_window", "bool"),
)


def _parse_env_bool(name: str, raw: str) -> bool:
    normalized = raw.strip().lower()
    if normalized in {"1", "true", "yes", "y", "on"}:
        return True
    if normalized in {"0", "false", "no", "n", "off"}:
        return False
    raise ValueError(f"{name} must be boolean-like")


def _parse_runtime_env_value(name: str, raw: str, kind: str) -> Any:
    if kind == "bool":
        return _parse_env_bool(name, raw)
    if kind == "regime_mode":
        normalized = raw.strip().lower().replace("-", "_")
        if normalized == "legacy":
            return "legacy"
        if normalized in {"dual", "dual_leg", "tiered", "cash_cap"}:
            return "dual_leg"
        raise ValueError(f"{name} must be legacy or dual_leg")
    if kind == "int":
        value = int(raw)
        if value < 1:
            raise ValueError(f"{name} must be >= 1")
        return value
    value = float(raw)
    if kind == "float":
        return value
    if kind == "ratio":
        if not 0.0 <= value <= 1.0:
            raise ValueError(f"{name} must be between 0 and 1")
        return value
    if kind == "positive_float":
        if value <= 0.0:
            raise ValueError(f"{name} must be > 0")
        return value
    raise ValueError(f"unsupported runtime env parser: {kind}")


def _load_combo_runtime_overrides() -> dict[str, Any]:
    overrides: dict[str, Any] = {}
    for env_name, config_key, kind in COMBO_RUNTIME_ENV_OVERRIDES:
        raw = os.getenv(env_name)
        if raw is None or not raw.strip():
            continue
        overrides[config_key] = _parse_runtime_env_value(env_name, raw, kind)
    return overrides


@dataclass(frozen=True)
class StrategyEvaluationResult:
    decision: StrategyDecision
    account_metrics: Mapping[str, Any] = field(default_factory=dict)
    metadata: Mapping[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class LoadedStrategyRuntime:
    entrypoint: StrategyEntrypoint
    runtime_adapter: StrategyRuntimeAdapter
    runtime_overrides: Mapping[str, Any] = field(default_factory=dict)
    merged_runtime_config: Mapping[str, Any] = field(default_factory=dict)
    local_artifact_candidates: tuple[Path, ...] = ()

    @property
    def profile(self) -> str:
        return self.entrypoint.manifest.profile

    @property
    def trend_pool_size(self) -> int:
        return int(self.merged_runtime_config.get("trend_pool_size", DEFAULT_TREND_POOL_SIZE))

    @property
    def artifact_contract(self) -> dict[str, Any]:
        contract = resolve_strategy_artifact_contract(self.runtime_adapter)
        return {
            "version": str(
                contract.snapshot_contract_version
                or self.merged_runtime_config.get("artifact_contract_version", "")
            ),
            "max_age_days": int(self.merged_runtime_config.get("artifact_max_age_days", 45)),
            "acceptable_modes": tuple(self.merged_runtime_config.get("artifact_acceptable_modes", ())),
            "requires_artifacts": bool(contract.requires_snapshot_artifacts),
            "requires_manifest": bool(contract.requires_snapshot_manifest_path),
            "config_source_policy": str(contract.config_source_policy),
            "default_local_candidates": tuple(str(path) for path in self.local_artifact_candidates),
        }

    @property
    def effective_runtime_config(self) -> dict[str, Any]:
        config = dict(self.merged_runtime_config)
        config.update(self.runtime_overrides)
        return config

    @property
    def default_local_artifact_path(self) -> Path:
        if self.local_artifact_candidates:
            return self.local_artifact_candidates[0]
        return DEFAULT_LOCAL_TREND_POOL_ARTIFACT

    def compute_account_metrics(
        self,
        runtime_trend_universe,
        balances,
        prices,
        u_total,
        fuel_val,
    ) -> dict[str, float]:
        trend_value = sum(float(balances[symbol]) * float(prices[symbol]) for symbol in runtime_trend_universe)
        dca_value = float(balances["BTCUSDT"]) * float(prices["BTCUSDT"])
        total_equity = float(u_total) + float(fuel_val) + trend_value + dca_value
        return {
            "cash_usdt": float(u_total),
            "trend_value": trend_value,
            "dca_value": dca_value,
            "total_equity": total_equity,
        }

    def build_portfolio_snapshot(
        self,
        *,
        account_metrics: Mapping[str, Any],
        balances: Mapping[str, Any] | None,
        prices: Mapping[str, Any],
        trend_universe_symbols: tuple[str, ...],
        as_of: datetime,
        portfolio_risk_symbols: tuple[str, ...] | None = None,
    ) -> PortfolioSnapshot:
        if not isinstance(account_metrics, Mapping):
            raise ValueError("portfolio_snapshot.account_metrics must be a mapping")
        if not isinstance(balances, Mapping) or not balances:
            raise ValueError("portfolio_snapshot.balances must be a complete non-empty mapping")
        if not isinstance(prices, Mapping):
            raise ValueError("portfolio_snapshot.prices must be a mapping")

        def finite_number(value: Any, *, label: str, minimum: float | None = None) -> float:
            if isinstance(value, bool):
                raise ValueError(f"portfolio_snapshot.{label} must be a finite number")
            try:
                normalized = float(value)
            except (TypeError, ValueError):
                raise ValueError(f"portfolio_snapshot.{label} must be a finite number") from None
            if not math.isfinite(normalized) or (minimum is not None and normalized < minimum):
                raise ValueError(f"portfolio_snapshot.{label} is invalid")
            return normalized

        total_equity = finite_number(account_metrics.get("total_equity"), label="total_equity", minimum=0.0)
        if total_equity <= 0.0:
            raise ValueError("portfolio_snapshot.total_equity must be positive")
        cash_balance = finite_number(account_metrics.get("cash_usdt"), label="cash_usdt", minimum=0.0)
        trend_value = finite_number(account_metrics.get("trend_value"), label="trend_value", minimum=0.0)
        dca_value = finite_number(account_metrics.get("dca_value"), label="dca_value", minimum=0.0)

        raw_symbols = tuple(
            portfolio_risk_symbols
            or ("BTCUSDT",) + tuple(str(symbol) for symbol in trend_universe_symbols)
        )
        risk_symbols = tuple(dict.fromkeys(symbol for symbol in raw_symbols if symbol != "USDT"))
        if not risk_symbols:
            raise ValueError("portfolio_snapshot.risk_assets must be non-empty")
        positions: list[Position] = []
        balances_map = dict(balances or {})
        observed_risk_value = 0.0
        for symbol in risk_symbols:
            if symbol not in balances_map:
                raise ValueError(f"portfolio_snapshot.balances missing {symbol}")
            if symbol not in prices:
                raise ValueError(f"portfolio_snapshot.prices missing {symbol}")
            quantity = finite_number(balances_map[symbol], label=f"balance.{symbol}", minimum=0.0)
            last_price = finite_number(prices[symbol], label=f"price.{symbol}")
            if last_price <= 0.0:
                raise ValueError(f"portfolio_snapshot.price.{symbol} must be positive")
            market_value = quantity * last_price
            if not math.isfinite(market_value) or market_value < 0.0:
                raise ValueError(f"portfolio_snapshot.market_value.{symbol} is invalid")
            observed_risk_value += market_value
            if quantity > 0.0:
                positions.append(Position(symbol=symbol, quantity=quantity, market_value=market_value))
        if not math.isclose(
            cash_balance + observed_risk_value,
            total_equity,
            rel_tol=1e-9,
            abs_tol=1e-6,
        ):
            raise ValueError("portfolio_snapshot.equity_reconciliation is invalid")
        observed_effective_exposure = observed_risk_value / total_equity
        return PortfolioSnapshot(
            as_of=as_of,
            total_equity=total_equity,
            buying_power=cash_balance,
            cash_balance=cash_balance,
            positions=tuple(positions),
            metadata={
                "account_metrics": dict(account_metrics),
                "cash_available_for_trading": cash_balance,
                "trend_value": trend_value,
                "dca_value": dca_value,
                "risk_asset_symbols": risk_symbols,
                "observed_risk_asset_value": observed_risk_value,
                "observed_effective_exposure": observed_effective_exposure,
            },
        )

    def evaluate(
        self,
        *,
        prices,
        trend_indicators,
        btc_snapshot,
        account_metrics,
        trend_universe_symbols,
        state,
        translator: Callable[..., str],
        balances: Mapping[str, Any] | None = None,
        portfolio_trend_universe_symbols: tuple[str, ...] | None = None,
        now_utc=None,
        allow_new_trend_entries: bool = True,
        allow_rotation_refresh: bool = True,
        get_symbol_trade_state_fn: Callable[..., Any] | None = None,
        set_symbol_trade_state_fn: Callable[..., Any] | None = None,
        mandate_provenance: Mapping[str, Any] | None = None,
        candidate_risk_identity: CandidateRiskIdentity | None = None,
        risk_authority: Any | None = None,
        runtime_target: Any | None = None,
        trend_pool_contract: Mapping[str, Any] | None = None,
        execution_mode: str | None = None,
        portfolio_risk_symbols: tuple[str, ...] | None = None,
    ) -> StrategyEvaluationResult:
        runtime_config = dict(self.runtime_overrides)
        runtime_config.update(
            {
                "translator": translator,
                "allow_new_trend_entries": bool(allow_new_trend_entries),
                "allow_rotation_refresh": bool(allow_rotation_refresh),
                "now_utc": now_utc,
            }
        )
        if get_symbol_trade_state_fn is not None:
            runtime_config["get_symbol_trade_state_fn"] = get_symbol_trade_state_fn
        if set_symbol_trade_state_fn is not None:
            runtime_config["set_symbol_trade_state_fn"] = set_symbol_trade_state_fn
        runtime_now = now_utc or datetime.now(timezone.utc)
        portfolio_symbols = (
            tuple(trend_universe_symbols)
            if portfolio_trend_universe_symbols is None
            else tuple(portfolio_trend_universe_symbols)
        )
        try:
            portfolio_snapshot = self.build_portfolio_snapshot(
                account_metrics=account_metrics,
                balances=balances,
                prices=prices,
                trend_universe_symbols=portfolio_symbols,
                as_of=runtime_now,
                portfolio_risk_symbols=portfolio_risk_symbols,
            )
        except ValueError as exc:
            # Preserve the existing CryptoStrategies → QPK gate so malformed
            # account inputs produce a redacted REJECT assessment and zero
            # execution, without inventing a zero-valued exposure snapshot.
            def invalid_snapshot_number(value: Any) -> float | None:
                try:
                    parsed = float(value)
                except (TypeError, ValueError):
                    return None
                return parsed if math.isfinite(parsed) else None

            portfolio_snapshot = PortfolioSnapshot(
                as_of=runtime_now,
                total_equity=invalid_snapshot_number(account_metrics.get("total_equity")),
                buying_power=invalid_snapshot_number(account_metrics.get("cash_usdt")),
                cash_balance=invalid_snapshot_number(account_metrics.get("cash_usdt")),
                positions=(),
                metadata={
                    "snapshot_validation_errors": (str(exc),),
                    "observed_effective_exposure": None,
                },
            )
        risk_material_errors: list[str] = []
        if execution_mode == "live" and risk_authority is None:
            # Live callers cannot inject a mandate/candidate pair directly.
            mandate_provenance = None
            candidate_risk_identity = None
            risk_material_errors.append("live_authority_material_missing")
        if risk_authority is not None:
            # The live entrypoint owns these values.  Caller-supplied mandate
            # and candidate objects cannot override a loaded authority file.
            mandate_provenance = None
            candidate_risk_identity = None
            try:
                if runtime_target is None:
                    raise LiveRiskAuthorityError("live risk authority configuration invalid: runtime target is missing")
                if not isinstance(trend_pool_contract, Mapping):
                    raise LiveRiskAuthorityError("live risk authority configuration invalid: trend pool contract is missing")
                mandate_provenance, candidate_risk_identity = bind_live_risk_authority(
                    risk_authority,
                    runtime_target=runtime_target,
                    input_material={
                        "prices": prices,
                        "balances": balances,
                        "account_metrics": account_metrics,
                        "trend_indicators": trend_indicators,
                        "btc_snapshot": btc_snapshot,
                        "state": state,
                        "universe": {
                            "strategy_trend": tuple(trend_universe_symbols),
                            "portfolio_trend": tuple(portfolio_symbols),
                        },
                        "execution_controls": {
                            "allow_new_trend_entries": bool(allow_new_trend_entries),
                            "allow_rotation_refresh": bool(allow_rotation_refresh),
                        },
                        "budget_observation": {
                            "managed_usdt": account_metrics.get("cash_usdt"),
                            "total_equity": portfolio_snapshot.total_equity,
                            "observed_effective_exposure": portfolio_snapshot.metadata.get(
                                "observed_effective_exposure"
                            ),
                        },
                        "trend_pool_contract": trend_pool_contract,
                    },
                    config_sha256=config_sha256(self.effective_runtime_config),
                    now_utc=runtime_now,
                )
            except (LiveRiskAuthorityError, ValueError):
                risk_material_errors.append("invalid_live_risk_authority")
        if candidate_risk_identity is not None and not isinstance(candidate_risk_identity, CandidateRiskIdentity):
            risk_material_errors.append("invalid_candidate_identity")
        if (
            isinstance(candidate_risk_identity, CandidateRiskIdentity)
            and candidate_risk_identity.strategy_profile != self.profile
        ):
            risk_material_errors.append("candidate_identity_strategy_profile_mismatch")
        if execution_mode == "live" and (
            not isinstance(mandate_provenance, Mapping)
            or mandate_provenance.get("authority_scope") != "LIVE"
        ):
            risk_material_errors.append("live_requires_live_mandate")
        candidate_for_context = (
            candidate_risk_identity
            if isinstance(candidate_risk_identity, CandidateRiskIdentity) and not risk_material_errors
            else None
        )
        from quant_platform_kit.strategy_lifecycle.live_equity import stamp_consecutive_losses_on_snapshot

        portfolio_snapshot = stamp_consecutive_losses_on_snapshot(
            portfolio_snapshot,
            strategy_profile=self.profile,
            domain="crypto",
            logger=getattr(self, "logger", None),
        )
        # Keep fuel and other account risk assets in the complete snapshot and
        # exposure metadata, while presenting only strategy-managed positions
        # to the pinned strategy stop resolver.  CryptoStrategies treats every
        # non-BTC snapshot position as a trend holding; BNB is an account fuel
        # asset and has no trend stop inputs.
        managed_symbols = {"BTCUSDT", *portfolio_symbols}
        managed_positions = tuple(
            position
            for position in portfolio_snapshot.positions
            if position.symbol in managed_symbols
        )
        if managed_positions != portfolio_snapshot.positions:
            portfolio_snapshot = replace(
                portfolio_snapshot,
                positions=managed_positions,
                metadata=dict(portfolio_snapshot.metadata),
            )
        evaluation_inputs = build_strategy_evaluation_inputs(
            available_inputs=self.runtime_adapter.available_inputs,
            market_inputs={
                "market_prices": prices,
                "derived_indicators": trend_indicators,
                "benchmark_snapshot": btc_snapshot,
                "universe_snapshot": tuple(trend_universe_symbols),
            },
            portfolio_snapshot=portfolio_snapshot,
        )
        ctx = build_strategy_context_from_available_inputs(
            entrypoint=self.entrypoint,
            runtime_adapter=self.runtime_adapter,
            as_of=runtime_now,
            available_inputs=evaluation_inputs,
            state=state,
            runtime_config=runtime_config,
            capabilities={"platform": BINANCE_PLATFORM},
        )
        artifacts = {"trend_pool_contract": self.artifact_contract}
        if isinstance(mandate_provenance, Mapping):
            artifacts["mandate_provenance"] = dict(mandate_provenance)
        if candidate_for_context is not None:
            artifacts["candidate_risk_identity"] = candidate_for_context
        ctx = StrategyContext(
            as_of=ctx.as_of,
            market_data=ctx.market_data,
            portfolio=ctx.portfolio,
            state=ctx.state,
            runtime_config=ctx.runtime_config,
            capabilities=ctx.capabilities,
            artifacts=artifacts,
        )
        decision = self.entrypoint.evaluate(ctx)
        return StrategyEvaluationResult(
            decision=decision,
            account_metrics=dict(account_metrics),
            metadata={
                "strategy_profile": self.profile,
                "strategy_display_name": resolve_research_strategy_metadata(
                    self.profile,
                    platform_id=BINANCE_PLATFORM,
                ).display_name,
            },
        )


def _build_loaded_strategy_runtime(entrypoint: StrategyEntrypoint) -> LoadedStrategyRuntime:
    runtime_adapter = get_platform_runtime_adapter(
        entrypoint.manifest.profile,
        platform_id=BINANCE_PLATFORM,
    )
    merged_runtime_config = dict(entrypoint.manifest.default_config)
    runtime_overrides: dict[str, Any] = {}
    if entrypoint.manifest.profile == "crypto_equity_combo":
        runtime_overrides.update(_load_combo_runtime_overrides())
    local_artifact_candidates = tuple(
        Path(path) for path in tp_get_default_live_pool_candidates(DEFAULT_LOCAL_TREND_POOL_ARTIFACT)
    )
    return LoadedStrategyRuntime(
        entrypoint=entrypoint,
        runtime_adapter=runtime_adapter,
        runtime_overrides=runtime_overrides,
        merged_runtime_config=merged_runtime_config,
        local_artifact_candidates=local_artifact_candidates,
    )


def load_strategy_runtime(raw_profile: str | None, *, runtime_target=None) -> LoadedStrategyRuntime:
    """Load an execution-eligible runtime through the platform policy gate."""

    entrypoint = (load_strategy_entrypoint_for_profile(raw_profile) if runtime_target is None
                  else load_strategy_entrypoint_for_profile(raw_profile, runtime_target=runtime_target))
    return _build_loaded_strategy_runtime(entrypoint)


def load_research_only_strategy_runtime(profile: str) -> LoadedStrategyRuntime:
    """Load a catalog runtime solely for local replay or import-safe utilities.

    This deliberately bypasses the runtime rollout allowlist, but it does not
    produce a RuntimeTarget or make a profile execution eligible.  The live
    configuration path still uses :func:`load_strategy_runtime` and therefore
    remains fail-closed when the allowlist is empty.
    """

    try:
        return _build_loaded_strategy_runtime(load_research_strategy_entrypoint_for_profile(profile))
    except (KeyError, TypeError, ValueError):
        raise ValueError(f"Unknown research-only strategy profile: {profile!r}") from None