""" 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 []) ], }