app/database/services.py

"""
Services — orchestrate repositories + business logic.
Loaded once at startup; CSV/JSON files are seeded into PostgreSQL here.
After this module runs, all queries go to PostgreSQL — never to CSVs.
"""
import os
import re
import glob
import csv
import json
import logging
from datetime import datetime
from typing import Optional

from sqlalchemy.orm import Session
from .repositories import (
    HistoricalPriceRepo, LiveMarketRepo, MarketNewsRepo, TechnicalIndicatorRepo
)
from . import models

logger = logging.getLogger(__name__)

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA_DIR = os.path.join(BASE_DIR, "data")
HISTORICAL_DIR = os.path.join(DATA_DIR, "simulation_historical_data")
PRICE_DIR = os.path.join(DATA_DIR, "simulation_price_data_July_1-Aug_30")
NEWS_DIR = os.path.join(DATA_DIR, "simulation_news_data_July_1-Aug_30")


# ─── Helpers ──────────────────────────────────────────────────────────────────

def _extract_symbol(filename: str) -> str:
    name = os.path.basename(filename).replace(".csv", "").replace(".json", "")
    for prefix in ["simulated_"]:
        if name.startswith(prefix):
            name = name[len(prefix):]
    for suffix in ["_2026_historical", "_historical", "_price_data", "_live"]:
        if name.endswith(suffix):
            name = name[: -len(suffix)]
    match = re.match(r"^[A-Za-z]+", name)
    return match.group(0).upper() if match else name.upper()


def _parse_dt(raw: str) -> Optional[datetime]:
    """Parse various timestamp formats to datetime."""
    if not raw:
        return None
    raw = raw.strip()
    for fmt in (
        "%Y-%m-%dT%H:%M:%SZ",
        "%Y-%m-%dT%H:%M:%S",
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%d",
    ):
        try:
            return datetime.strptime(raw, fmt)
        except ValueError:
            continue
    return None


def _normalize_alphavantage_ts(ts: str) -> Optional[datetime]:
    """Convert AlphaVantage compact format 20260701T062006 → datetime."""
    if not ts:
        return None
    ts = ts.strip()
    if len(ts) == 15 and "T" in ts:
        try:
            return datetime.strptime(ts, "%Y%m%dT%H%M%S")
        except ValueError:
            pass
    return _parse_dt(ts)


# ─── CSV Loaders ──────────────────────────────────────────────────────────────

def _load_historical_csvs_to_pg(db: Session) -> int:
    total = 0
    if not os.path.exists(HISTORICAL_DIR):
        return 0
    for filepath in glob.glob(os.path.join(HISTORICAL_DIR, "*.csv")):
        sym = _extract_symbol(filepath)
        rows = []
        closes = []
        raw_rows = []
        with open(filepath, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                try:
                    ts = _parse_dt(row.get("timestamp", row.get("date", "")))
                    if not ts:
                        continue
                    raw_rows.append({
                        "symbol": sym,
                        "timestamp": ts,
                        "open": round(float(row["open"]), 4),
                        "high": round(float(row["high"]), 4),
                        "low": round(float(row["low"]), 4),
                        "close": round(float(row["close"]), 4),
                        "adjusted_close": round(float(row.get("adjusted_close") or row["close"]), 4),
                        "volume": int(float(row.get("volume", 0))),
                        "dividend_amount": float(row.get("dividend_amount", 0)),
                        "split_coefficient": float(row.get("split_coefficient", 1.0)),
                    })
                    closes.append(float(row["close"]))
                except (ValueError, KeyError):
                    continue

        # Compute MA20, MA50, RSI
        for i, r in enumerate(raw_rows):
            r["ma20"] = round(sum(closes[max(0, i - 19):i + 1]) / min(20, i + 1), 2) if i >= 19 else None
            r["ma50"] = round(sum(closes[max(0, i - 49):i + 1]) / min(50, i + 1), 2) if i >= 49 else None
            if i >= 14:
                gains, losses = [], []
                for j in range(i - 13, i + 1):
                    prev_c = closes[j - 1] if j > 0 else closes[j]
                    diff = closes[j] - prev_c
                    (gains if diff > 0 else losses).append(abs(diff))
                ag = sum(gains) / 14.0
                al = sum(losses) / 14.0
                r["rsi"] = 100.0 if al == 0 else round(100.0 - (100.0 / (1.0 + ag / al)), 1)
            else:
                r["rsi"] = None

        inserted = HistoricalPriceRepo.upsert(db, raw_rows)
        total += inserted
        logger.info(f"  Historical {sym}: inserted {inserted}/{len(raw_rows)} rows")
    return total


def _load_live_csvs_to_pg(db: Session) -> int:
    total = 0
    if not os.path.exists(PRICE_DIR):
        return 0
    for filepath in glob.glob(os.path.join(PRICE_DIR, "*.csv")):
        sym = _extract_symbol(filepath)
        rows = []
        with open(filepath, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                try:
                    ts = _parse_dt(row.get("timestamp", ""))
                    if not ts:
                        continue
                    o = float(row["open"])
                    c = float(row["close"])
                    rows.append({
                        "symbol": sym,
                        "timestamp": ts,
                        "open": round(o, 4),
                        "high": round(float(row["high"]), 4),
                        "low": round(float(row["low"]), 4),
                        "close": round(c, 4),
                        "volume": int(float(row.get("volume", 0))),
                        "vwap": round((o + c) / 2.0, 4),
                    })
                except (ValueError, KeyError):
                    continue
        inserted = LiveMarketRepo.upsert(db, rows)
        total += inserted
        logger.info(f"  Live {sym}: inserted {inserted}/{len(rows)} rows")
    return total


def _load_news_jsons_to_pg(db: Session) -> int:
    total = 0
    if not os.path.exists(NEWS_DIR):
        return 0
    id_counter = 1
    for filepath in glob.glob(os.path.join(NEWS_DIR, "*.json")):
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                raw = json.load(f)

            items = []
            if isinstance(raw, list):
                items = raw
            elif isinstance(raw, dict):
                if "feed" in raw and isinstance(raw["feed"], list):
                    items = raw["feed"]
                else:
                    for val in raw.values():
                        if isinstance(val, list):
                            items.extend(val)

            items = items[:50]
            rows = []
            for item in items:
                ts_raw = item.get("time_published", "")
                pub_at = _normalize_alphavantage_ts(ts_raw) if ts_raw else datetime.utcnow()

                ticker_sents = item.get("ticker_sentiment", [])
                first_ts = ticker_sents[0] if ticker_sents else {}
                raw_score = float(first_ts.get("ticker_sentiment_score", 0) or 0)
                rel_score = float(first_ts.get("relevance_score", 0.8) or 0.8)
                sent_label = first_ts.get("ticker_sentiment_label", "")
                sentiment = (
                    "bullish" if "bullish" in sent_label.lower()
                    else "bearish" if "bearish" in sent_label.lower()
                    else "neutral"
                )
                conf = min(0.99, max(0.65, (abs(raw_score) * 0.5 + rel_score * 0.5)))
                ov_score = raw_score

                topics_raw = item.get("topics", [])
                topics = [
                    {"topic": t.get("topic", "General"), "relevance_score": float(t.get("relevance_score", 0))}
                    for t in topics_raw
                ]
                tickers = [
                    {
                        "ticker": ts.get("ticker", ""),
                        "relevance_score": float(ts.get("relevance_score", 0) or 0),
                        "sentiment_score": float(ts.get("ticker_sentiment_score", 0) or 0),
                        "sentiment_label": ts.get("ticker_sentiment_label", ""),
                    }
                    for ts in ticker_sents
                    if ts.get("ticker")
                ]

                news_id = f"news-{filepath[-16:-5].replace('/', '-')}-{id_counter}"
                rows.append({
                    "news_id": news_id,
                    "headline": item.get("title", "Market Update"),
                    "summary": item.get("summary", item.get("title", "")),
                    "source": item.get("source", "MarketWatch"),
                    "url": item.get("url", ""),
                    "published_at": pub_at,
                    "sentiment": sentiment,
                    "confidence_score": round(conf, 3),
                    "overall_sentiment_score": round(ov_score, 3),
                    "importance_score": round(rel_score, 3),
                    "is_breaking": conf >= 0.90,
                    "topics": topics,
                    "ticker_sentiments": tickers,
                })
                id_counter += 1

            inserted = MarketNewsRepo.bulk_insert(db, rows)
            total += inserted
            logger.info(f"  News {os.path.basename(filepath)}: inserted {inserted}/{len(rows)} items")
        except Exception as e:
            logger.warning(f"Error loading news {filepath}: {e}")
    return total


# ─── Master Startup Seeder ─────────────────────────────────────────────────────

def seed_postgres(db: Session) -> dict:
    """
    Called once at FastAPI startup.
    Loads all CSV/JSON data into PostgreSQL.
    Idempotent — skips tables if data already exists for instant startup.
    """
    logger.info("=== PostgreSQL seed starting ===")
    
    # Check existing row counts to avoid redundant startup loading
    has_hist = db.query(models.HistoricalPrice.id).first() is not None if hasattr(models, 'HistoricalPrice') else False
    has_live = db.query(models.LiveMarketData.id).first() is not None if hasattr(models, 'LiveMarketData') else False
    has_news = db.query(models.MarketNews.id).first() is not None if hasattr(models, 'MarketNews') else False

    hist = 0 if has_hist else _load_historical_csvs_to_pg(db)
    live = 0 if has_live else _load_live_csvs_to_pg(db)
    news = 0 if has_news else _load_news_jsons_to_pg(db)

    logger.info(f"=== Seed complete: hist={hist} live={live} news={news} ===")
    return {"historical": hist, "live": live, "news": news}


# ─── Query Services (used by routes) ──────────────────────────────────────────

class MarketDataService:
    @staticmethod
    def get_market_summary(db: Session) -> list[dict]:
        return LiveMarketRepo.get_market_summary(db)

    @staticmethod
    def get_historical(db: Session, symbol: str, limit: int = 500) -> list[dict]:
        rows = HistoricalPriceRepo.get_by_symbol(db, symbol, limit=limit)
        return [
            {
                "date": r.timestamp.strftime("%Y-%m-%d"),
                "open": r.open, "high": r.high, "low": r.low, "close": r.close,
                "volume": r.volume, "ma20": r.ma20, "ma50": r.ma50, "rsi": r.rsi,
            }
            for r in rows
        ]

    @staticmethod
    def get_live_ticks(db: Session, symbol: str, limit: int = 1000) -> list[dict]:
        rows = LiveMarketRepo.get_by_symbol(db, symbol, limit=limit)
        return [
            {
                "date": r.timestamp.isoformat(),
                "open": r.open, "high": r.high, "low": r.low, "close": r.close,
                "volume": r.volume, "vwap": r.vwap,
            }
            for r in rows
        ]

    @staticmethod
    def get_latest_price(db: Session, symbol: str) -> Optional[dict]:
        r = LiveMarketRepo.get_latest(db, symbol) or HistoricalPriceRepo.get_latest(db, symbol)
        if not r:
            return None
        return {"symbol": symbol.upper(), "close": r.close, "timestamp": r.timestamp.isoformat()}


class NewsService:
    @staticmethod
    def get_feed(db: Session, symbol: Optional[str] = None, topic: Optional[str] = None,
                 sentiment: Optional[str] = None, limit: int = 50, offset: int = 0) -> list[dict]:
        rows = MarketNewsRepo.get_feed(db, symbol=symbol, topic=topic, sentiment=sentiment, limit=limit, offset=offset)
        return [NewsService._serialize(r) for r in rows]

    @staticmethod
    def get_trending_topics(db: Session, hours: int = 24) -> list[dict]:
        return MarketNewsRepo.get_trending_topics(db, hours=hours)

    @staticmethod
    def get_ticker_sentiment_summary(db: Session, hours: int = 24) -> list[dict]:
        return MarketNewsRepo.get_ticker_sentiment_summary(db, hours=hours)

    @staticmethod
    def get_news_velocity(db: Session, hours: int = 6) -> list[dict]:
        return MarketNewsRepo.get_news_velocity(db, hours=hours)

    @staticmethod
    def get_sector_sentiment(db: Session, hours: int = 24) -> list[dict]:
        return MarketNewsRepo.get_sector_sentiment(db, hours=hours)

    @staticmethod
    def get_breaking_news(db: Session) -> list[dict]:
        rows = MarketNewsRepo.get_breaking_news(db)
        return [NewsService._serialize(r) for r in rows]

    @staticmethod
    def get_by_id(db: Session, news_id: str) -> Optional[dict]:
        row = MarketNewsRepo.get_by_id(db, news_id)
        return NewsService._serialize(row) if row else None

    @staticmethod
    def _serialize(r) -> dict:
        if r is None:
            return {}
        return {
            "id": r.news_id,
            "db_id": r.id,
            "headline": r.headline,
            "summary": r.summary,
            "source": r.source,
            "url": r.url,
            "published_at": r.published_at.isoformat() if r.published_at else None,
            "sentiment": r.sentiment,
            "confidence_score": r.confidence_score,
            "overall_sentiment_score": r.overall_sentiment_score,
            "importance_score": r.importance_score,
            "is_breaking": r.is_breaking,
            "topics": [{"topic": t.topic, "relevance_score": t.relevance_score} for t in (r.topics or [])],
            "tickers": [
                {
                    "ticker": ts.ticker,
                    "relevance_score": ts.relevance_score,
                    "sentiment_score": ts.sentiment_score,
                    "sentiment_label": ts.sentiment_label,
                }
                for ts in (r.ticker_sentiments or [])
            ],
        }