spb/crypto-trade-bot

1"""Strategy SDK.23Lifecycle (identical in backtest, paper and live engines):4  1. `features(df)`      — causal feature computation on closed bars (row t may only use rows <= t)5  2. `target(feats)`     — target exposure in [0, 1] (spot: no shorting unless the exchange account supports it)6                           decided AT THE CLOSE of bar t and executable no earlier than the OPEN of bar t+17  3. `uncertainty(feats)`— optional per-bar confidence/uncertainty (used for sizing & reporting)8  4. risk gating + execution are done by the engine, never by the strategy.910Causality is enforced by tests (`tests/test_no_lookahead.py`): for every registered strategy, targets computed11on a truncated history must equal targets computed on the full history, bar for bar.12"""1314from __future__ import annotations1516from abc import ABC, abstractmethod17from dataclasses import dataclass, field18from typing import Any1920import numpy as np21import pandas as pd22from pydantic import BaseModel232425class ParamSpec(BaseModel):26    name: str27    type: str = "int"  # int | float | bool | choice28    default: Any29    min: float | None = None30    max: float | None = None31    step: float | None = None32    choices: list[Any] | None = None33    description: str = ""343536class StrategyMeta(BaseModel):37    name: str38    label: str39    family: str40    hypothesis: str41    universe_note: str = "Kraken spot pairs in the configured universe"42    experimental: bool = True43    allows_short: bool = False44    min_bars: int = 5045    params: list[ParamSpec] = []46    failure_modes: list[str] = []47    references: list[str] = []484950@dataclass51class StrategyResult:52    features: pd.DataFrame53    target: pd.Series  # exposure fraction in [0,1] (or [-1,1] if shorting allowed), index-aligned with features54    uncertainty: pd.Series | None = None55    diagnostics: dict[str, Any] = field(default_factory=dict)565758class Strategy(ABC):59    meta: StrategyMeta6061    def __init__(self, params: dict[str, Any] | None = None) -> None:62        self.params = self.validate_params(params or {})6364    @classmethod65    def validate_params(cls, params: dict[str, Any]) -> dict[str, Any]:66        out: dict[str, Any] = {}67        for spec in cls.meta.params:68            v = params.get(spec.name, spec.default)69            if spec.type == "int":70                v = int(v)71            elif spec.type == "float":72                v = float(v)73            elif spec.type == "bool":74                v = bool(v) if not isinstance(v, str) else v.lower() in ("1", "true", "yes")75            elif spec.type == "choice" and spec.choices is not None and v not in spec.choices:76                raise ValueError(f"{spec.name}: {v!r} not in {spec.choices}")77            if spec.min is not None and isinstance(v, int | float) and v < spec.min:78                raise ValueError(f"{spec.name}: {v} < min {spec.min}")79            if spec.max is not None and isinstance(v, int | float) and v > spec.max:80                raise ValueError(f"{spec.name}: {v} > max {spec.max}")81            out[spec.name] = v82        unknown = set(params) - {s.name for s in cls.meta.params}83        if unknown:84            raise ValueError(f"unknown params: {sorted(unknown)}")85        return out8687    @abstractmethod88    def features(self, df: pd.DataFrame) -> pd.DataFrame: ...8990    @abstractmethod91    def target(self, feats: pd.DataFrame) -> pd.Series: ...9293    def uncertainty(self, feats: pd.DataFrame) -> pd.Series | None:94        return None9596    def run(self, df: pd.DataFrame) -> StrategyResult:97        feats = self.features(df)98        tgt = self.target(feats).astype(float)99        lo = -1.0 if self.meta.allows_short else 0.0100        tgt = tgt.clip(lo, 1.0).fillna(0.0)101        return StrategyResult(features=feats, target=tgt, uncertainty=self.uncertainty(feats))102103104def log_returns(close: pd.Series) -> pd.Series:105    return np.log(close).diff()106107108def vol_target_scale(ret: pd.Series, lookback: int, target_annual_vol: float, bars_per_year: float, cap: float = 1.0) -> pd.Series:109    """Position scale so that realized vol ≈ target; uses only trailing returns (shifted to be causal at bar close)."""110    rv = ret.rolling(lookback, min_periods=max(5, lookback // 2)).std(ddof=1) * np.sqrt(bars_per_year)111    scale = (target_annual_vol / rv.replace(0, np.nan)).clip(upper=cap)112    return scale.fillna(0.0)113