stock-oracle/app/core/config.py

"""
Configuration settings for Stock Oracle API
"""

from typing import List, Union, Dict
from pydantic import AnyHttpUrl, field_validator
try:
    from pydantic_settings import BaseSettings
except ImportError:
    from pydantic import BaseSettings
import os
import json
from dotenv import load_dotenv

load_dotenv()

class Settings(BaseSettings):
    # Application
    APP_NAME: str = "Stock Oracle"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = True
    ENVIRONMENT: str = "development"
    
    # API
    API_PREFIX: str = "/api/v1"
    
    # CORS
    BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
    
    @field_validator("BACKEND_CORS_ORIGINS", mode="before")
    def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
        if isinstance(v, str) and not v.startswith("["):
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, (list, str)):
            return v
        raise ValueError(v)
    
    # Database
    DATABASE_URL: str = os.getenv(
        "DATABASE_URL",
        "postgresql+asyncpg://stockoracle:stockoracle2024@localhost:15433/stock_oracle"
    )
    DATABASE_ECHO: bool = False
    DB_POOL_SIZE: int = int(os.getenv("DB_POOL_SIZE", "20"))
    DB_MAX_OVERFLOW: int = int(os.getenv("DB_MAX_OVERFLOW", "30"))
    
    # Redis
    REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:16380/0")
    CACHE_TTL: int = 3600  # 1 hour default
    
    # SEC Settings
    SEC_EMAIL: str = os.getenv("SEC_EMAIL", "example@example.com")
    SEC_DATA_REFRESH_HOURS: int = 24
    SEC_DATA_START_YEAR: int = 1994  # SEC EDGAR data available from 1994
    
    # Security
    SECRET_KEY: str = os.getenv("SECRET_KEY", "development-secret-key-change-in-production")
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    # ETF Scraper integration
    ETF_SCRAPER_TICKERS: Union[List[str], str] = ['MTUM']  # e.g., ["MTUM", "QTUM"] or "MTUM,QTUM"
    @field_validator("ETF_SCRAPER_TICKERS", mode="before")
    def parse_scraper_tickers(cls, v):
        if isinstance(v, str):
            if not v:
                return []
            return [t.strip().upper() for t in v.split(",") if t.strip()]
        if isinstance(v, list):
            return [str(t).strip().upper() for t in v]
        return []

    # Manual ETF inception/start dates (ISO YYYY-MM-DD), used to short-circuit pre-launch requests
    # Accepts either a JSON string or a comma-separated "TICKER:YYYY-MM-DD" list via env
    ETF_START_DATES: Dict[str, str] = {"MTUM": "2013-04-16"}

    @field_validator("ETF_START_DATES", mode="before")
    def parse_etf_start_dates(cls, v):
        # Examples:
        #   "{\"MTUM\": \"2013-04-16\", \"QQQ\": \"1999-03-10\"}"
        #   "MTUM:2013-04-16,QQQ:1999-03-10"
        if isinstance(v, str):
            v = v.strip()
            if not v:
                return {}
            try:
                data = json.loads(v)
                if isinstance(data, dict):
                    return {str(k).strip().upper(): str(val).strip() for k, val in data.items() if str(val).strip()}
            except Exception:
                pass
            items = {}
            for part in v.split(','):
                if not part.strip():
                    continue
                if ':' in part:
                    k, val = part.split(':', 1)
                    k = k.strip().upper()
                    val = val.strip()
                    if k and val:
                        items[k] = val
            return items
        if isinstance(v, dict):
            return {str(k).strip().upper(): str(val).strip() for k, val in v.items() if str(val).strip()}
        return {}
    
    # Alpaca Market Data
    # Alpha Vantage (Earnings Surprise)
    ALPHA_VANTAGE_API_KEY: str = os.getenv("ALPHA_VANTAGE_API_KEY", "")

    ALPACA_API_KEY: str = os.getenv("ALPACA_API_KEY", "")
    ALPACA_SECRET_KEY: str = os.getenv("ALPACA_SECRET_KEY", "")
    ALPACA_BASE_URL: str = os.getenv("ALPACA_BASE_URL", "https://data.alpaca.markets")

    # Server
    API_PORT: int = int(os.getenv("API_PORT", "18000"))
    DB_PORT: int = int(os.getenv("DB_PORT", "15432"))
    REDIS_PORT: int = int(os.getenv("REDIS_PORT", "16380"))
    
    # Migration
    MIGRATION_API_KEY: str = os.getenv("MIGRATION_API_KEY", "migration-key-change-in-production")
    ALLOW_MIGRATION: bool = os.getenv("ALLOW_MIGRATION", "True").lower() == "true"

    # Overlay (Phase 5)
    YOUTUBE_API_KEY: str = os.getenv("YOUTUBE_API_KEY", "")
    GOOGLE_TRENDS_ENABLED: bool = os.getenv("GOOGLE_TRENDS_ENABLED", "false").lower() == "true"
    OVERLAY_ENABLED: bool = os.getenv("OVERLAY_ENABLED", "true").lower() == "true"
    OVERLAY_STALE_HOURS: int = int(os.getenv("OVERLAY_STALE_HOURS", "8"))
    
    class Config:
        case_sensitive = True
        env_file = ".env"

settings = Settings()

# Warn about default secrets in non-development environments
import logging as _logging
_config_logger = _logging.getLogger(__name__)
if settings.ENVIRONMENT != "development":
    if settings.SECRET_KEY == "development-secret-key-change-in-production":
        _config_logger.warning("SECRET_KEY is using the default value! Set a secure SECRET_KEY for production.")
    if settings.MIGRATION_API_KEY == "migration-key-change-in-production":
        _config_logger.warning("MIGRATION_API_KEY is using the default value! Set a secure MIGRATION_API_KEY for production.")