File size: 5,146 Bytes
d53dc44 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
contracts.py — Cross-module typed interfaces.
Every inter-module call MUST use these types. No raw dicts between modules.
"""
from __future__ import annotations
import datetime
from typing import Literal
import pandas as pd
from pydantic import BaseModel, Field, field_validator
# ── Exceptions ───────────────────────────────────────────────────────────────
class InsufficientDataError(Exception):
"""Raised when fewer than n_bars are available for a symbol/timeframe."""
class ConfigError(Exception):
"""Raised when configuration validation fails."""
class BrokerError(Exception):
"""Raised on unrecoverable broker API errors."""
class SafeModeError(Exception):
"""Raised when an operation is attempted while SAFE MODE is active."""
# ── Signal Contracts ─────────────────────────────────────────────────────────
class TechnicalScore(BaseModel):
"""Output of signals/technical.py for a single symbol at a point in time."""
model_config = {"arbitrary_types_allowed": True}
symbol: str
timestamp: datetime.datetime
n_signals_computed: int = 0
score: float = Field(ge=-1.0, le=1.0)
sub_signals: dict[str, float] = Field(default_factory=dict)
skip_reason: str | None = None
@field_validator("score", mode="before")
@classmethod
def force_zero_if_insufficient(cls, v: float, info) -> float:
n = info.data.get("n_signals_computed", 0)
if n < 3:
return 0.0
return v
class SentimentScore(BaseModel):
"""Output of signals/sentiment.py for a single symbol."""
symbol: str
score: float = Field(ge=-1.0, le=1.0)
cached_at: datetime.datetime
source_count: int = 0
stale: bool = False
@field_validator("stale", mode="before")
@classmethod
def compute_stale(cls, v: bool, info) -> bool:
cached = info.data.get("cached_at")
if cached is None:
return True
now = datetime.datetime.now(datetime.timezone.utc)
if cached.tzinfo is None:
cached = cached.replace(tzinfo=datetime.timezone.utc)
delta = (now - cached).total_seconds()
return delta > 35 * 60 # stale if > 35 minutes
class FinalScore(BaseModel):
"""Combined signal output from signals/combiner.py."""
symbol: str
timestamp: datetime.datetime
technical: float
sentiment: float
final: float
decision: Literal["BUY", "SELL", "HOLD"]
skip_reason: str | None = None
class PriceTarget(BaseModel):
"""Predicted price targets for dashboard display."""
symbol: str
current_price: float
predicted_direction: Literal["UP", "DOWN", "NEUTRAL"]
entry_price: float | None = None
stop_loss_price: float | None = None
take_profit_price: float | None = None
expected_profit_usd: float = 0.0
confidence: float = Field(ge=0.0, le=1.0, default=0.0)
# ── Order Contracts ──────────────────────────────────────────────────────────
class OrderRequest(BaseModel):
"""Internal contract for placing orders."""
symbol: str
side: Literal["buy", "sell"]
qty: float
limit_price: float
stop_price: float | None = None
tp_price: float | None = None
trail_percent: float | None = None
is_fractional: bool = False
dry_run: bool = False
class LinkedOrderGroup(BaseModel):
"""Tracks entry + stop + TP legs for fractional orders."""
symbol: str
entry_order_id: str
stop_order_id: str | None = None
tp_order_id: str | None = None
entry_filled: bool = False
stop_submitted: bool = False
tp_submitted: bool = False
is_fractional: bool = False
entry_price: float | None = None
orphaned: bool = False
# ── Data Contract ────────────────────────────────────────────────────────────
# get_bars(symbol, timeframe, n_bars, adjusted=True) → pd.DataFrame
# Columns: timestamp (UTC index), open, high, low, close, volume, vwap, adjusted
# Raises InsufficientDataError if fewer than n_bars available.
BAR_COLUMNS = ["open", "high", "low", "close", "volume", "vwap", "adjusted"]
def validate_bar_dataframe(df: pd.DataFrame, symbol: str, n_bars: int) -> pd.DataFrame:
"""Validate that a bars DataFrame matches the contract."""
if len(df) < n_bars:
raise InsufficientDataError(
f"{symbol}: requested {n_bars} bars, got {len(df)}"
)
missing = [c for c in BAR_COLUMNS if c not in df.columns]
if missing:
raise ValueError(f"{symbol}: missing columns {missing}")
return df
|