read.markets/app/config.py

"""Runtime configuration — environment via Pydantic Settings + TOML-loaded data tables.

Settings come from .env / process env. The TOML files (default.toml, portfolio.toml)
define *what to track* — they're declarative content, not config knobs, so they
stay separate from the settings model.
"""
from __future__ import annotations

import tomllib
from functools import lru_cache
from pathlib import Path
from typing import Any

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict


CONFIG_DIR = Path(__file__).resolve().parent.parent / "config"


class Settings(BaseSettings):
    """All runtime knobs. Read from process env, .env not needed in-container
    because compose injects vars directly; .env is supported for local dev."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # Database
    DATABASE_URL: str = "mysql+aiomysql://cassandra:changeme@db:3306/cassandra"
    # Redis: ephemeral pie storage during /api/analyze + batch buffer for
    # ticker_universe additions. No persistence — see compose service.
    REDIS_URL: str = "redis://redis:6379/0"

    # API keys (mirror prototype .env names)
    API_KEY: str = ""              # Trading 212 key
    SECRET_KEY: str = ""           # Trading 212 secret
    FRED_API_KEY: str = ""
    OPENROUTER_API_KEY: str = ""

    # App
    CASSANDRA_TOKEN: str = ""
    CASSANDRA_PORT: int = 8000
    # Signing key for session cookies. Generate with:
    #   python -c "import secrets; print(secrets.token_urlsafe(32))"
    # Falls back to CASSANDRA_TOKEN if unset (acceptable for single-host dev).
    CASSANDRA_SESSION_SECRET: str = ""
    # Set to false (or 0/no) to disable /signup after the first account is
    # created. Phase A leaves this open so the operator can self-onboard.
    CASSANDRA_SIGNUP_ENABLED: bool = True

    # SMTP for email OTP verification. If SMTP_SERVER is empty, OTP codes
    # are written to stdout instead of sent — convenient for local dev.
    SMTP_SERVER: str = ""
    SMTP_PORT: int = 587
    SMTP_USER: str = ""
    SMTP_PASSWORD: str = ""
    SMTP_USE_TLS: bool = True
    SMTP_FROM: str = ""    # Defaults to SMTP_USER if blank
    CASSANDRA_BASE_CURRENCY: str = "GBP"
    CASSANDRA_ANCHOR_DATE: str = ""
    CASSANDRA_MOCK: bool = False

    # AI log — provider abstraction with fallback chain.
    # `LLM_PROVIDER` is the primary; `LLM_FALLBACK` kicks in if the primary
    # raises (after its own internal retries). Set LLM_FALLBACK="" to
    # disable the fallback.
    LLM_PROVIDER: str = "deepseek"
    LLM_FALLBACK: str = "openrouter"

    # DeepSeek-direct (cheaper, primary).
    DEEPSEEK_API_KEY: str = ""
    DEEPSEEK_URL: str = "https://api.deepseek.com/chat/completions"
    DEEPSEEK_MODEL: str = "deepseek-v4-flash"

    # OpenRouter (fallback, also a valid primary).
    OPENROUTER_MODEL: str = "deepseek/deepseek-v4-flash"
    OPENROUTER_MONTHLY_CAP_USD: float = 20.0
    # Tone axis. PRO was dropped in PROMPT_VERSION 6 (audience pivot to
    # young investors); legacy values are silently mapped to INTERMEDIATE
    # by app.services.openrouter._resolve_tone.
    CASSANDRA_TONE: str = "INTERMEDIATE"      # NOVICE | INTERMEDIATE
    CASSANDRA_ANALYSIS: str = "SPECULATIVE"   # DRY | SPECULATIVE

    # Config file locations (overridable for tests)
    BASELINE_TOML: Path = Field(default_factory=lambda: CONFIG_DIR / "default.toml")
    PORTFOLIO_TOML: Path = Field(default_factory=lambda: CONFIG_DIR / "portfolio.toml")


@lru_cache
def get_settings() -> Settings:
    return Settings()


# --- TOML data tables --------------------------------------------------------


def _merge_toml(*paths: Path) -> dict[str, Any]:
    """Read TOML files in order; later ones override earlier at the top level
    (with shallow dict-merge for nested tables)."""
    out: dict[str, Any] = {}
    for path in paths:
        if not path.exists():
            continue
        with path.open("rb") as f:
            data = tomllib.load(f)
        for k, v in data.items():
            if isinstance(v, dict) and isinstance(out.get(k), dict):
                out[k].update(v)
            else:
                out[k] = v
    return out


def load_groups(*paths: Path) -> dict[str, list[tuple[str, str, str]]]:
    """[(symbol, label, note), ...] per group name."""
    data = _merge_toml(*paths)
    out: dict[str, list[tuple[str, str, str]]] = {}
    for name, items in (data.get("groups") or {}).items():
        out[name] = [
            (it["symbol"], it.get("label", it["symbol"]), it.get("note", ""))
            for it in items
        ]
    return out


def load_feeds(*paths: Path) -> dict[str, list[tuple[str, str]]]:
    """[(name, url), ...] per category name."""
    data = _merge_toml(*paths)
    out: dict[str, list[tuple[str, str]]] = {}
    for cat, items in (data.get("feeds") or {}).items():
        out[cat] = [(it["name"], it["url"]) for it in items]
    return out


def load_presets(*paths: Path) -> dict[str, list[str]]:
    """Keyword presets for news filtering."""
    data = _merge_toml(*paths)
    presets = (data.get("news") or {}).get("presets") or {}
    return {name: list(kw) for name, kw in presets.items()}


def load_all() -> tuple[dict, dict, dict]:
    """Shortcut: groups, feeds, presets using the configured TOML paths."""
    s = get_settings()
    return (
        load_groups(s.BASELINE_TOML, s.PORTFOLIO_TOML),
        load_feeds(s.BASELINE_TOML, s.PORTFOLIO_TOML),
        load_presets(s.BASELINE_TOML, s.PORTFOLIO_TOML),
    )