zwitschi
1df4b11aef
CI / lint-test-build (push) Failing after 1m7s
- Introduced new HTML templates for the dashboard, metrics, overview, and backtesting functionalities. - Implemented partial templates for metrics, overview, audit, controls, and charts to enhance modularity. - Updated the Jinja2 template resolution logic to support different deployment environments. - Added a health check template to display the service status. - Included a test suite to verify the template resolution logic. - Updated `pyproject.toml` to include new HTML templates in the package data.
153 lines
4.6 KiB
Python
153 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import deque
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from statistics import fmean, pstdev
|
|
from typing import Literal
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class StatArbExperimentConfig:
|
|
pair_a: str
|
|
pair_b: str
|
|
lookback_window: int = 120
|
|
entry_zscore: float = 2.0
|
|
exit_zscore: float = 0.5
|
|
max_holding_seconds: float = 900.0
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class StatArbSignal:
|
|
action: Literal[
|
|
"warmup",
|
|
"hold",
|
|
"enter_long_spread",
|
|
"enter_short_spread",
|
|
"exit_position",
|
|
]
|
|
observed_at: datetime
|
|
spread: float
|
|
zscore: float | None
|
|
position: Literal["long", "short", "flat"]
|
|
|
|
|
|
class StatArbExperiment:
|
|
"""Simple mean-reversion experiment scaffold behind feature flags."""
|
|
|
|
def __init__(self, config: StatArbExperimentConfig) -> None:
|
|
if config.lookback_window < 2:
|
|
raise ValueError("lookback_window must be >= 2")
|
|
if config.entry_zscore <= 0.0:
|
|
raise ValueError("entry_zscore must be > 0")
|
|
if config.exit_zscore < 0.0:
|
|
raise ValueError("exit_zscore must be >= 0")
|
|
if config.entry_zscore <= config.exit_zscore:
|
|
raise ValueError("entry_zscore must be > exit_zscore")
|
|
if config.max_holding_seconds <= 0.0:
|
|
raise ValueError("max_holding_seconds must be > 0")
|
|
|
|
self._config = config
|
|
self._spreads: deque[float] = deque(maxlen=config.lookback_window)
|
|
self._position: Literal["long", "short", "flat"] = "flat"
|
|
self._position_opened_at: datetime | None = None
|
|
|
|
@property
|
|
def config(self) -> StatArbExperimentConfig:
|
|
return self._config
|
|
|
|
def reset(self) -> None:
|
|
self._spreads.clear()
|
|
self._position = "flat"
|
|
self._position_opened_at = None
|
|
|
|
def observe(
|
|
self,
|
|
*,
|
|
price_a: float,
|
|
price_b: float,
|
|
observed_at: datetime,
|
|
) -> StatArbSignal:
|
|
if price_a <= 0.0 or price_b <= 0.0:
|
|
raise ValueError("prices must be > 0")
|
|
|
|
at = observed_at.astimezone(UTC)
|
|
spread = price_a - price_b
|
|
self._spreads.append(spread)
|
|
|
|
if len(self._spreads) < self._config.lookback_window:
|
|
return StatArbSignal(
|
|
action="warmup",
|
|
observed_at=at,
|
|
spread=spread,
|
|
zscore=None,
|
|
position=self._position,
|
|
)
|
|
|
|
mean_spread = fmean(self._spreads)
|
|
std_spread = pstdev(self._spreads)
|
|
if std_spread == 0.0:
|
|
return StatArbSignal(
|
|
action="hold",
|
|
observed_at=at,
|
|
spread=spread,
|
|
zscore=0.0,
|
|
position=self._position,
|
|
)
|
|
|
|
zscore = (spread - mean_spread) / std_spread
|
|
|
|
if self._position == "flat":
|
|
if zscore >= self._config.entry_zscore:
|
|
self._position = "short"
|
|
self._position_opened_at = at
|
|
return StatArbSignal(
|
|
action="enter_short_spread",
|
|
observed_at=at,
|
|
spread=spread,
|
|
zscore=zscore,
|
|
position=self._position,
|
|
)
|
|
if zscore <= -self._config.entry_zscore:
|
|
self._position = "long"
|
|
self._position_opened_at = at
|
|
return StatArbSignal(
|
|
action="enter_long_spread",
|
|
observed_at=at,
|
|
spread=spread,
|
|
zscore=zscore,
|
|
position=self._position,
|
|
)
|
|
return StatArbSignal(
|
|
action="hold",
|
|
observed_at=at,
|
|
spread=spread,
|
|
zscore=zscore,
|
|
position=self._position,
|
|
)
|
|
|
|
assert self._position_opened_at is not None
|
|
held_seconds = (at - self._position_opened_at).total_seconds()
|
|
should_exit = abs(zscore) <= self._config.exit_zscore
|
|
if held_seconds >= self._config.max_holding_seconds:
|
|
should_exit = True
|
|
|
|
if should_exit:
|
|
self._position = "flat"
|
|
self._position_opened_at = None
|
|
return StatArbSignal(
|
|
action="exit_position",
|
|
observed_at=at,
|
|
spread=spread,
|
|
zscore=zscore,
|
|
position=self._position,
|
|
)
|
|
|
|
return StatArbSignal(
|
|
action="hold",
|
|
observed_at=at,
|
|
spread=spread,
|
|
zscore=zscore,
|
|
position=self._position,
|
|
)
|