ai/tools/csv_tools.py

"""
CSV Data Access Tools for Mirai AI Copilot.
Reads directly from backend/app/data directories.
"""

import os
import glob
import json
import csv
import re
from typing import Dict, Any, List

BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
DATA_DIR = os.path.join(BASE_DIR, "app", "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")

def normalize_symbol(symbol: str) -> str:
    if not symbol:
        return "AAPL"
    sym = symbol.upper().strip()
    if sym in ["GOOGL", "ALPHABET", "GOOGLE"]:
        return "GOOG"
    return sym

def extract_symbol(query: str, default_symbol: str = "AAPL") -> str:
    if not query:
        return normalize_symbol(default_symbol)
    
    query_upper = query.upper()
    known_symbols = ["GOOGL", "GOOG", "AAPL", "TSLA", "MSFT", "IBM", "WMT", "UL", "NVDA", "AMZN", "META"]
    for sym in known_symbols:
        if re.search(r'\b' + sym + r'\b', query_upper):
            return normalize_symbol(sym)
    return normalize_symbol(default_symbol)

def getHistoricalData(symbol: str, period: str = "3M") -> List[Dict[str, Any]]:
    sym = normalize_symbol(symbol)
    bars = []
    
    files = glob.glob(os.path.join(HISTORICAL_DIR, f"*{sym}*.csv"))
    filepath = files[0] if files else None

    if filepath and os.path.exists(filepath):
        with open(filepath, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                try:
                    bars.append({
                        "date": row.get("timestamp", "").strip(),
                        "open": float(row.get("open", 0)),
                        "high": float(row.get("high", 0)),
                        "low": float(row.get("low", 0)),
                        "close": float(row.get("close", 0)),
                        "volume": int(row.get("volume", 0))
                    })
                except (ValueError, KeyError):
                    continue
    return bars[-63:] if len(bars) > 63 else bars

def getLivePrice(symbol: str) -> Dict[str, Any]:
    sym = normalize_symbol(symbol)
    ticks = []
    
    files = glob.glob(os.path.join(PRICE_DIR, f"*{sym}*.csv"))
    filepath = files[0] if files else None

    if filepath and os.path.exists(filepath):
        with open(filepath, "r", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                try:
                    ticks.append({
                        "timestamp": row.get("timestamp", "").strip(),
                        "open": float(row.get("open", 0)),
                        "high": float(row.get("high", 0)),
                        "low": float(row.get("low", 0)),
                        "close": float(row.get("close", 0)),
                        "volume": int(row.get("volume", 0))
                    })
                except (ValueError, KeyError):
                    continue

    if ticks:
        latest = ticks[-1]
        prev = ticks[-2] if len(ticks) >= 2 else latest
        chg = round(latest["close"] - prev["close"], 2)
        pct = round((chg / prev["close"] * 100), 2) if prev["close"] else 0.0
        return {
            "symbol": sym,
            "price": latest["close"],
            "change": chg,
            "changePercent": pct,
            "volume": latest["volume"],
            "bid": round(latest["close"] * 0.9995, 2),
            "ask": round(latest["close"] * 1.0005, 2)
        }
    
    return {"symbol": sym, "price": 194.50, "change": -0.68, "changePercent": -0.35, "volume": 124500, "bid": 194.49, "ask": 194.51}

def getMarketNews(symbol: str = None, limit: int = 10) -> List[Dict[str, Any]]:
    items = []
    sym = normalize_symbol(symbol) if symbol else ""

    if os.path.exists(NEWS_DIR):
        for filepath in sorted(glob.glob(os.path.join(NEWS_DIR, "*.json")), reverse=True):
            try:
                with open(filepath, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    
                    news_lists = []
                    if isinstance(data, dict):
                        for date_key in sorted(data.keys(), reverse=True):
                            news_lists.extend(data[date_key])
                    elif isinstance(data, list):
                        news_lists = data
                    
                    for item in news_lists:
                        tickers = []
                        if "ticker_sentiment" in item and isinstance(item["ticker_sentiment"], list):
                            tickers = [ts.get("ticker", "").upper() for ts in item["ticker_sentiment"]]
                        elif "symbols" in item:
                            tickers = [s.upper() for s in item.get("symbols", [])]

                        # Check if news matches symbol or return recent news if no symbol specified
                        is_match = False
                        if not symbol:
                            is_match = True
                        else:
                            if sym in tickers or (sym == "GOOG" and "GOOGL" in tickers):
                                is_match = True

                        if is_match:
                            sentiment_label = "Neutral"
                            if "ticker_sentiment" in item and isinstance(item["ticker_sentiment"], list):
                                for ts in item["ticker_sentiment"]:
                                    if ts.get("ticker", "").upper() in [sym, "GOOG", "GOOGL"]:
                                        sentiment_label = ts.get("ticker_sentiment_label", "Neutral")
                                        break
                            
                            items.append({
                                "title": item.get("title", "Market Update"),
                                "summary": item.get("summary", item.get("title", "")),
                                "source": item.get("source", "Financial News"),
                                "time_published": item.get("time_published", ""),
                                "sentiment": sentiment_label,
                                "tickers": tickers
                            })
                            if len(items) >= limit:
                                return items
            except Exception:
                continue
    return items[:limit]