src/config.py

"""Central configuration for the Backtest Lab.

Everything that a maintainer might want to tune -- repo ids, provider chains,
asset universe, rate limits, guardrails -- lives here as data, not code.
"""

from __future__ import annotations

import os
from dataclasses import dataclass, field

# --------------------------------------------------------------------------
# Repos
# --------------------------------------------------------------------------

ORG = "The-Bit-Trading-Company"

# The shared signal store lives under the org (the company-branded data asset).
STORE_REPO = os.environ.get("BIT_STORE_REPO", f"{ORG}/bit-signal-store")
STORE_REPO_TYPE = "dataset"

# The Space itself. Gradio Spaces under an org require a paid Team/Enterprise
# plan, so the app is hosted under the owner's PRO personal namespace.
# See DECISIONS.md (D-001).
SPACE_REPO = os.environ.get("BIT_SPACE_REPO", "Bit-Trading-Company/bit-backtest-lab")

MANIFEST_PATH = "manifest.json"
MANIFEST_SCHEMA_VERSION = 1

# Bumped whenever a change to inference or storage semantics invalidates
# previously-written signal slices.
INFERENCE_VERSION = "1.0.0"
PLACEHOLDER_VERSION = "PLACEHOLDER"

# --------------------------------------------------------------------------
# Assets & timeframes
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class Asset:
    """One tradable symbol, with the per-provider symbol spellings it needs."""

    slug: str  # canonical id used in store paths, e.g. "BTC-USD"
    display: str
    kind: str  # "crypto" | "equity"
    ccxt_symbol: str | None = None
    yahoo_symbol: str | None = None
    stooq_symbol: str | None = None
    tiingo_symbol: str | None = None


ASSETS: dict[str, Asset] = {
    a.slug: a
    for a in [
        Asset("BTC-USD", "Bitcoin", "crypto", ccxt_symbol="BTC/USDT", yahoo_symbol="BTC-USD"),
        Asset("ETH-USD", "Ethereum", "crypto", ccxt_symbol="ETH/USDT", yahoo_symbol="ETH-USD"),
        Asset("SOL-USD", "Solana", "crypto", ccxt_symbol="SOL/USDT", yahoo_symbol="SOL-USD"),
        Asset("SPY", "S&P 500 ETF", "equity", yahoo_symbol="SPY",
              stooq_symbol="spy.us", tiingo_symbol="SPY"),
        Asset("QQQ", "Nasdaq 100 ETF", "equity", yahoo_symbol="QQQ",
              stooq_symbol="qqq.us", tiingo_symbol="QQQ"),
        Asset("NVDA", "NVIDIA", "equity", yahoo_symbol="NVDA",
              stooq_symbol="nvda.us", tiingo_symbol="NVDA"),
    ]
}


@dataclass(frozen=True)
class Timeframe:
    slug: str
    pandas_freq: str
    minutes: int
    bars_per_year: float
    ccxt_tf: str | None = None
    yahoo_interval: str | None = None
    # Provider-imposed history depth, in days. None = no practical limit.
    # These are honest coverage boundaries, not errors (see data.py).
    yahoo_max_days: int | None = None


TIMEFRAMES: dict[str, Timeframe] = {
    t.slug: t
    for t in [
        Timeframe("1d", "D", 1440, 365.0, ccxt_tf="1d", yahoo_interval="1d"),
        Timeframe("1h", "h", 60, 365.0 * 24, ccxt_tf="1h", yahoo_interval="1h",
                  yahoo_max_days=730),
        Timeframe("15m", "15min", 15, 365.0 * 24 * 4, ccxt_tf="15m",
                  yahoo_interval="15m", yahoo_max_days=60),
    ]
}

# Equities only trade during market hours, so a calendar year holds far fewer
# bars than the wall-clock math above. Annualisation uses these instead.
EQUITY_BARS_PER_YEAR = {"1d": 252.0, "1h": 252.0 * 6.5, "15m": 252.0 * 26.0}


def bars_per_year(asset_slug: str, tf_slug: str) -> float:
    """Annualisation factor for Sharpe/CAGR, respecting market calendars."""
    asset = ASSETS.get(asset_slug)
    if asset is not None and asset.kind == "equity":
        return EQUITY_BARS_PER_YEAR[tf_slug]
    return TIMEFRAMES[tf_slug].bars_per_year


# --------------------------------------------------------------------------
# Provider chain (config, not code -- data.py walks these in order)
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class ProviderSpec:
    name: str
    kinds: tuple[str, ...]
    # Minimum seconds between calls, and backoff schedule on failure.
    min_interval_s: float = 0.25
    max_retries: int = 4
    backoff_base_s: float = 1.5
    requires_env: str | None = None


PROVIDER_CHAIN: tuple[ProviderSpec, ...] = (
    ProviderSpec("binance", ("crypto",), min_interval_s=0.10),
    ProviderSpec("coinbase", ("crypto",), min_interval_s=0.35),
    ProviderSpec("yfinance", ("equity",), min_interval_s=1.20),
    ProviderSpec("stooq", ("equity",), min_interval_s=1.00),
    ProviderSpec("tiingo", ("equity",), min_interval_s=0.60, requires_env="TIINGO_KEY"),
)


def providers_for(kind: str) -> list[ProviderSpec]:
    """Ordered, currently-usable providers for an asset kind."""
    out = []
    for p in PROVIDER_CHAIN:
        if kind not in p.kinds:
            continue
        if p.requires_env and not os.environ.get(p.requires_env):
            continue
        out.append(p)
    return out


# --------------------------------------------------------------------------
# Models
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class ModelSpec:
    slug: str  # store path segment
    model_id: str  # HF model id
    family: str  # adapter family
    display: str
    context_len: int = 512
    quantile_levels: tuple[float, ...] = (0.1, 0.5, 0.9)


SEED_MODELS: dict[str, ModelSpec] = {
    m.slug: m
    for m in [
        # Chronos-Bolt: the fast encoder-decoder family. All four sizes share one
        # adapter, so comparing them isolates model capacity from everything else.
        ModelSpec("chronos-bolt-tiny", "amazon/chronos-bolt-tiny", "chronos",
                  "Chronos-Bolt Tiny", context_len=512),
        ModelSpec("chronos-bolt-mini", "amazon/chronos-bolt-mini", "chronos",
                  "Chronos-Bolt Mini", context_len=512),
        ModelSpec("chronos-bolt-small", "amazon/chronos-bolt-small", "chronos",
                  "Chronos-Bolt Small", context_len=512),
        ModelSpec("chronos-bolt-base", "amazon/chronos-bolt-base", "chronos",
                  "Chronos-Bolt Base", context_len=512),
        # Original Chronos (T5-based, sampling rather than direct quantiles).
        ModelSpec("chronos-t5-small", "amazon/chronos-t5-small", "chronos",
                  "Chronos T5 Small", context_len=512),

        # Chronos-2. Loads through the same adapter and the already-pinned
        # chronos-forecasting 2.3.1, but `predict_quantiles` returns a list of
        # per-item tensors rather than one stacked tensor -- see
        # `ChronosAdapter._to_array`. Seedable, so it costs GPU quota on the
        # next seed run; drop it from SEEDABLE_MODELS if that is not wanted yet.
        ModelSpec("chronos-2", "amazon/chronos-2", "chronos",
                  "Chronos-2", context_len=512),

        # Naive baselines, deliberately first-class. A forecasting model that
        # cannot beat "tomorrow looks like today" is not worth deploying, and
        # the leaderboard should make that impossible to miss.
        ModelSpec("baseline-naive", "baseline/naive", "baseline",
                  "Baseline · Random walk", context_len=128),
        ModelSpec("baseline-drift", "baseline/drift", "baseline",
                  "Baseline · Drift", context_len=128),
        ModelSpec("baseline-seasonal", "baseline/seasonal", "baseline",
                  "Baseline · Seasonal naive", context_len=128),

        # Registered but unseeded: the timesfm package is heavy and optional.
        ModelSpec("timesfm-2-500m", "google/timesfm-2.0-500m-pytorch", "timesfm",
                  "TimesFM 2.0 500M", context_len=512),
    ]
}

# Models the seed plan actually runs. TimesFM is excluded until its dependency
# is pinned in requirements.txt.
SEEDABLE_MODELS = tuple(k for k in SEED_MODELS if not k.startswith("timesfm"))

BASELINE_MODELS = tuple(k for k, v in SEED_MODELS.items() if v.family == "baseline")


def is_baseline(model_slug: str) -> bool:
    return model_slug in BASELINE_MODELS

# Adapter families a user may pick from in the "Add model" flow. Restricting to
# a fixed set is what keeps arbitrary model code from ever being executed.
ALLOWED_ADAPTER_FAMILIES = ("chronos", "timesfm", "baseline")


# --------------------------------------------------------------------------
# Guardrails for user-funded coverage extension
# --------------------------------------------------------------------------


@dataclass(frozen=True)
class ExtensionCaps:
    max_days: dict[str, int] = field(
        default_factory=lambda: {"1d": 730, "1h": 183, "15m": 62}
    )
    max_steps_per_run: int = 4000
    smoke_test_steps: int = 100


CAPS = ExtensionCaps()

# --------------------------------------------------------------------------
# Backtest defaults
# --------------------------------------------------------------------------

DEFAULT_INIT_CASH = 10_000.0
DEFAULT_COMMISSION_BPS = 10.0  # per side
DEFAULT_SLIPPAGE_BPS = 5.0
DEFAULT_HOLDOUT_MONTHS = 6

# In-process LRU sizing for parquet slices (Phase 3 perf target: <2s runs).
PARQUET_CACHE_SIZE = 64

DISCLAIMER = (
    "Backtested results are hypothetical, derived from historical data, and are "
    "not indicative of future results. Nothing here is investment advice. "
    "The Bit Trading Company is not a licensed investment adviser."
)