"""
Analyst & News Cache Builder
============================
Runs nightly via GitHub Actions. Fetches analyst consensus data and
news sentiment for every US-listed stock and upserts results into the
Supabase analyst_cache table.

Phase 1 — Analyst data via yf.Ticker.info        (~50 min for 6,500 tickers)
Phase 2 — News sentiment via Google News RSS      (~6 min, 20 workers)

Each phase writes to Supabase incrementally every UPSERT_EVERY records
so a partial run is never fully lost.
"""

import html
import os
import random
import re
import time
import urllib.parse
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, date, timedelta, timezone

import requests
import yfinance as yf

try:
    from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
    _VADER = SentimentIntensityAnalyzer()
except ImportError:
    _VADER = None

# ---------------------------------------------------------------------------
# PostgreSQL connection — reads secrets from /srv/api/.env on the VPS,
# or from environment variables when run locally.
# ---------------------------------------------------------------------------
import psycopg2
import psycopg2.extras
from dotenv import load_dotenv
load_dotenv("/srv/api/.env")

_DB_HOST = "127.0.0.1"
_DB_NAME = "verimund"
_DB_USER = "verimund_user"
_DB_PASS = os.environ.get("DB_PASS", "Shlevison2k17")

# TABLE_PREFIX is set by the cron job: "" for stable, "beta_" for beta channel.
_TABLE_PREFIX = os.environ.get("TABLE_PREFIX", "")


def _get_conn():
    return psycopg2.connect(host=_DB_HOST, dbname=_DB_NAME,
                            user=_DB_USER, password=_DB_PASS)


def _pg_upsert(table: str, rows: list[dict], conflict_col: str = "ticker") -> None:
    """Generic PostgreSQL upsert — INSERT ... ON CONFLICT DO UPDATE."""
    if not rows:
        return
    import math
    # Sanitize: NaN/Inf → None (PostgreSQL JSON can't handle these)
    def _clean(v):
        return None if isinstance(v, float) and not math.isfinite(v) else v

    cols = list(rows[0].keys())
    col_str  = ", ".join(f'"{c}"' for c in cols)
    vals_str = ", ".join(["%s"] * len(cols))
    upd_str  = ", ".join(f'"{c}" = EXCLUDED."{c}"' for c in cols if c != conflict_col)
    sql = (
        f'INSERT INTO "{table}" ({col_str}) VALUES ({vals_str}) '
        f'ON CONFLICT ("{conflict_col}") DO UPDATE SET {upd_str}'
    )
    data = [[_clean(row.get(c)) for c in cols] for row in rows]
    try:
        conn = _get_conn()
        with conn.cursor() as cur:
            cur.executemany(sql, data)
        conn.commit()
        conn.close()
        print(f"  Upserted {len(rows)} rows → {table}")
    except Exception as e:
        print(f"  [WARN] Upsert to {table} failed: {e}")

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_TICKER_RE    = re.compile(r'^[A-Z]{1,5}(-[A-Z]{1,2})?$')
_HTTP_HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}

INFO_BATCH   = 20
INFO_PAUSE   = 4.0
MAX_WORKERS  = 5
NEWS_WORKERS = 20
UPSERT_EVERY = 500

# ---------------------------------------------------------------------------
# Foreign / annual filers — companies that don't appear in EDGAR quarterly
# frames because they file Form 20-F (annual only).
# Split into two groups:
#   US_GAAP_ANNUAL  — file 20-F but report under US GAAP (same concept names,
#                     just annual data instead of quarterly)
#   IFRS_ANNUAL     — file 20-F under IFRS (different concept names, mapped
#                     to equivalent US GAAP fields below)
#   NO_XBRL         — no structured XBRL data; skipped gracefully
# ---------------------------------------------------------------------------

_EDGAR_UA = {"User-Agent": "StockScreener contact@investmenttool.com"}

US_GAAP_ANNUAL_FILERS = {
    "AEM":   "0000002809",   # Agnico Eagle Mines
    "ARM":   "0001973239",   # ARM Holdings
    "ASML":  "0000937966",   # ASML Holding
    "BABA":  "0001577552",   # Alibaba Group
    "BAM":   "0001937926",   # Brookfield Asset Management
    "BBVA":  "0000842180",   # Banco Bilbao Vizcaya
    "BHP":   "0000811809",   # BHP Group
    "HDB":   "0001144967",   # HDFC Bank
    "HTHIY": "0000047710",   # Hitachi
    "ING":   "0001039765",   # ING Groep
    "ITUB":  "0001132597",   # Itau Unibanco
    "LYG":   "0001160106",   # Lloyds Banking Group
    "PDD":   "0001737806",   # PDD Holdings
    "SAN":   "0000891478",   # Banco Santander
    "TM":    "0001094517",   # Toyota Motor
}

IFRS_ANNUAL_FILERS = {
    "AZN":   "0000901832",   # AstraZeneca
    "BCS":   "0000312069",   # Barclays
    "BP":    "0000313807",   # BP
    "GSK":   "0001131399",   # GSK
    "HSBC":  "0001089113",   # HSBC Holdings
    "NGG":   "0001004315",   # National Grid
    "NU":    "0001691493",   # Nu Holdings
    "NVS":   "0001114448",   # Novartis
    "RIO":   "0000863064",   # Rio Tinto plc
    "SHEL":  "0001306965",   # Shell
    "SPOT":  "0001639920",   # Spotify
    "UBS":   "0001610520",   # UBS Group
    "UL":    "0000217410",   # Unilever
}

# IFRS concept → internal field name mapping
IFRS_CONCEPT_MAP = {
    "Revenue":                                    "revenue",
    "ProfitLoss":                                 "net_income",
    "ProfitLossAttributableToOwnersOfParent":     "net_income",
    "OperatingIncomeLoss":                        "operating_income",
    "Assets":                                     "assets",
    "Equity":                                     "equity",
    "EquityAttributableToOwnersOfParent":         "equity",
    "CurrentAssets":                              "cur_assets",
    "CurrentLiabilities":                         "cur_liabilities",
    "CashAndCashEquivalents":                     "cash",
    "NoncurrentPortionOfLongtermBorrowings":      "lt_debt",
    "Borrowings":                                 "lt_debt",
    "AdjustmentsForDepreciationAndAmortisationExpense": "dna",
    "CashFlowsFromUsedInOperatingActivities":     "op_cf",
    "PurchaseOfPropertyPlantAndEquipment":        "capex",
}

# US GAAP annual concept → internal field name mapping (same as quarterly)
USGAAP_CONCEPT_MAP = {
    "Revenues":                                              "revenue",
    "RevenueFromContractWithCustomerExcludingAssessedTax":   "revenue",
    "SalesRevenueNet":                                       "revenue",
    "NetIncomeLoss":                                         "net_income",
    "OperatingIncomeLoss":                                   "operating_income",
    "Assets":                                                "assets",
    "StockholdersEquity":                                    "equity",
    "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest": "equity",
    "AssetsCurrent":                                         "cur_assets",
    "LiabilitiesCurrent":                                    "cur_liabilities",
    "CashAndCashEquivalentsAtCarryingValue":                 "cash",
    "CashCashEquivalentsAndShortTermInvestments":            "cash",
    "LongTermDebt":                                          "lt_debt",
    "LongTermDebtNoncurrent":                                "lt_debt",
    "LongTermDebtAndFinanceLeaseLiabilities":                "lt_debt",
    "FinanceLeaseLiabilityNoncurrent":                       "lt_debt",
    "CommonStockholdersEquity":                              "equity",
    "DepreciationDepletionAndAmortization":                  "dna",
    "DepreciationAndAmortization":                           "dna",
    "NetCashProvidedByUsedInOperatingActivities":            "op_cf",
    "PaymentsToAcquirePropertyPlantAndEquipment":            "capex",
    "CommonStockSharesOutstanding":                          "shares",
}


# ---------------------------------------------------------------------------
# Ticker fetching (NASDAQ Trader files — same source as the screener)
# ---------------------------------------------------------------------------

def _clean_ticker(symbol: str) -> str | None:
    t = symbol.strip().upper().replace(".", "-")
    return t if _TICKER_RE.match(t) else None


def fetch_tickers() -> list[str]:
    """Return all US common-stock tickers from NASDAQ Trader listing files."""
    tickers = []
    base    = "https://www.nasdaqtrader.com/dynamic/SymDir/"

    # NASDAQ-listed stocks
    try:
        resp = requests.get(base + "nasdaqlisted.txt", headers=_HTTP_HEADERS, timeout=20)
        resp.raise_for_status()
        for line in resp.text.splitlines()[1:]:
            if line.startswith("File Creation"):
                break
            parts = line.split("|")
            if len(parts) < 7:
                continue
            if parts[3].strip() == "Y" or parts[6].strip() == "Y":   # test/ETF
                continue
            t = _clean_ticker(parts[0].strip())
            if t:
                tickers.append(t)
        print(f"  NASDAQ listed : {sum(1 for _ in tickers)} tickers")
    except Exception as e:
        print(f"  [WARN] nasdaqlisted.txt: {e}")

    # NYSE / AMEX listed stocks
    try:
        resp = requests.get(base + "otherlisted.txt", headers=_HTTP_HEADERS, timeout=20)
        resp.raise_for_status()
        count = 0
        for line in resp.text.splitlines()[1:]:
            if line.startswith("File Creation"):
                break
            parts = line.split("|")
            if len(parts) < 7:
                continue
            if parts[2].strip() not in {"N", "A"}:   # NYSE=N, AMEX=A only
                continue
            if parts[4].strip() == "Y" or parts[6].strip() == "Y":   # ETF/test
                continue
            t = _clean_ticker(parts[0].strip())
            if t:
                tickers.append(t)
                count += 1
        print(f"  NYSE/AMEX     : {count} tickers")
    except Exception as e:
        print(f"  [WARN] otherlisted.txt: {e}")

    combined = list(dict.fromkeys(tickers))
    print(f"  Total         : {len(combined)} unique tickers\n")
    return combined


# ---------------------------------------------------------------------------
# Phase 1 — Analyst data
# ---------------------------------------------------------------------------

def _fetch_analyst(ticker: str) -> dict | None:
    """Fetch analyst consensus fields for one ticker via yf.Ticker.info."""
    time.sleep(random.uniform(0.0, 1.0))
    try:
        info = yf.Ticker(ticker).info
        if not info or not isinstance(info, dict) or len(info) < 3:
            return None

        try:
            rec_mean = float(info["recommendationMean"]) if info.get("recommendationMean") is not None else None
        except (TypeError, ValueError):
            rec_mean = None
        analyst_norm = (5.0 - rec_mean) / 4.0 if rec_mean is not None and 1 <= rec_mean <= 5 else None

        try:
            target = float(info["targetMeanPrice"]) if info.get("targetMeanPrice") is not None else None
        except (TypeError, ValueError):
            target = None

        price = info.get("currentPrice") or info.get("regularMarketPrice")
        try:
            price = float(price) if price is not None else None
        except (TypeError, ValueError):
            price = None

        analyst_upside = (target / price - 1.0) if target and price and price > 0 else None

        return {
            "ticker":         ticker,
            "pe_forward":     info.get("forwardPE"),
            "eps_forward":    info.get("forwardEps"),
            "analyst_norm":   analyst_norm,
            "analyst_upside": analyst_upside,
            "analyst_count":  info.get("numberOfAnalystOpinions"),
            "analyst_target": target,
            "recommendation": info.get("recommendationKey") or "N/A",
            "sector":         info.get("sector")   or None,
            "industry":       info.get("industry") or None,
            "dividend_yield": info.get("dividendYield"),
            "updated_at":     datetime.now(timezone.utc).isoformat(),
        }
    except Exception:
        return None


def run_analyst_phase(tickers: list[str]) -> None:
    print(f"Phase 1: Analyst data for {len(tickers)} tickers ...")
    total   = len(tickers)
    done    = 0
    pending: list[dict] = []
    start   = time.time()

    for batch_start in range(0, total, INFO_BATCH):
        batch = tickers[batch_start: batch_start + INFO_BATCH]
        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:
            futures = {ex.submit(_fetch_analyst, t): t for t in batch}
            for fut in as_completed(futures):
                done += 1
                result = fut.result()
                if result:
                    pending.append(result)
                elapsed = time.time() - start
                eta     = (elapsed / done) * (total - done) if done else 0
                print(
                    f"\r  {done}/{total} ({done / total * 100:.0f}%)"
                    f"  cached={len(pending)}  ETA={eta:.0f}s   ",
                    end="", flush=True,
                )

        if len(pending) >= UPSERT_EVERY:
            print()
            _upsert(pending)
            pending.clear()

        if batch_start + INFO_BATCH < total:
            time.sleep(INFO_PAUSE)

    if pending:
        print()
        _upsert(pending)

    print(f"\n  Phase 1 complete in {time.time() - start:.0f}s\n")


# ---------------------------------------------------------------------------
# Phase 2 — News sentiment (Google News RSS + VADER)
# ---------------------------------------------------------------------------

def _fetch_news_sentiment(ticker: str) -> dict | None:
    """Fetch Google News RSS headlines and compute VADER compound score."""
    if _VADER is None:
        return None
    try:
        query   = urllib.parse.quote(f"{ticker} stock")
        rss_url = (
            f"https://news.google.com/rss/search"
            f"?q={query}&hl=en-US&gl=US&ceid=US:en"
        )
        resp = requests.get(rss_url, headers=_HTTP_HEADERS, timeout=8)
        resp.raise_for_status()
        root   = ET.fromstring(resp.content)
        scores = []
        for item in root.findall(".//item")[:15]:
            title_el = item.find("title")
            if title_el is not None and title_el.text:
                title = html.unescape(
                    re.sub(r"<[^>]+>", " ", title_el.text)
                ).strip()
                scores.append(_VADER.polarity_scores(title)["compound"])
        sentiment = sum(scores) / len(scores) if scores else None
        return {
            "ticker":         ticker,
            "news_sentiment": sentiment,
            "updated_at":     datetime.now(timezone.utc).isoformat(),
        }
    except Exception:
        return None


def run_news_phase(tickers: list[str]) -> None:
    if _VADER is None:
        print("Phase 2: Skipped — vaderSentiment not installed\n")
        return

    print(f"Phase 2: News sentiment for {len(tickers)} tickers ...")
    total   = len(tickers)
    done    = 0
    pending: list[dict] = []
    start   = time.time()

    with ThreadPoolExecutor(max_workers=NEWS_WORKERS) as ex:
        futures = {ex.submit(_fetch_news_sentiment, t): t for t in tickers}
        for fut in as_completed(futures):
            done += 1
            result = fut.result()
            if result:
                pending.append(result)
            if done % 200 == 0:
                elapsed = time.time() - start
                eta     = (elapsed / done) * (total - done) if done else 0
                print(
                    f"\r  {done}/{total} ({done / total * 100:.0f}%)"
                    f"  ETA={eta:.0f}s   ",
                    end="", flush=True,
                )
            if len(pending) >= UPSERT_EVERY:
                _upsert(pending)
                pending.clear()

    if pending:
        _upsert(pending)

    print(f"\n  Phase 2 complete in {time.time() - start:.0f}s\n")


# ---------------------------------------------------------------------------
# Upsert helpers — write to local PostgreSQL
# ---------------------------------------------------------------------------

# Explicit column allow-lists keep extra fields (e.g. dividend_yield from
# yfinance) from causing "column does not exist" errors in PostgreSQL.
_ANALYST_COLS = (
    "ticker", "pe_forward", "eps_forward", "analyst_norm", "analyst_upside",
    "analyst_count", "analyst_target", "recommendation", "news_sentiment",
    "sector", "industry", "updated_at",
)
_FUND_COLS = (
    "ticker", "pe_trailing", "pb_ratio", "ev_ebitda", "eps_trailing",
    "revenue", "revenue_growth", "earnings_growth", "roe", "roa",
    "debt_to_equity", "total_debt", "total_cash", "book_value", "current_ratio",
    "profit_margin", "operating_margin", "fcf_yield", "dividend_yield",
    "market_cap", "shares_outstanding", "ev_revenue", "short_percent",
    "filer_type", "updated_at",
)


def _upsert(rows: list[dict]) -> None:
    """Upsert a batch of rows into analyst_cache."""
    filtered = [{c: r.get(c) for c in _ANALYST_COLS} for r in rows]
    _pg_upsert(f"{_TABLE_PREFIX}analyst_cache", filtered)


def _upsert_fundamentals(rows: list[dict]) -> None:
    """Upsert a batch of rows into fundamentals_cache."""
    filtered = [{c: r.get(c) for c in _FUND_COLS} for r in rows]
    _pg_upsert(f"{_TABLE_PREFIX}fundamentals_cache", filtered)


# ---------------------------------------------------------------------------
# Phase 3 helpers — EDGAR bulk frames (domestic quarterly filers)
# ---------------------------------------------------------------------------

_FRAME_CACHE: dict = {}


def _get_recent_quarters(n: int = 9) -> list[tuple[int, int]]:
    ref = date.today() - timedelta(days=45)
    y, q = ref.year, (ref.month - 1) // 3 + 1
    out = []
    for _ in range(n):
        out.append((y, q))
        q -= 1
        if q == 0:
            q, y = 4, y - 1
    return out


def _fetch_one_frame(concept: str, unit: str, period: str) -> dict[int, float]:
    key = (concept, unit, period)
    if key in _FRAME_CACHE:
        return _FRAME_CACHE[key]
    url = f"https://data.sec.gov/api/xbrl/frames/us-gaap/{concept}/{unit}/{period}.json"
    try:
        resp = requests.get(url, headers=_EDGAR_UA, timeout=20)
        if resp.status_code == 404:
            _FRAME_CACHE[key] = {}
            return {}
        resp.raise_for_status()
        result: dict[int, float] = {}
        for row in resp.json().get("data", []):
            try:
                if isinstance(row, dict):
                    cik, val = int(row["cik"]), float(row["val"])
                else:
                    if len(row) < 6:
                        continue
                    cik, val = int(row[1]), float(row[5])
                result[cik] = val
            except (KeyError, TypeError, ValueError):
                continue
        _FRAME_CACHE[key] = result
        time.sleep(0.15)
        return result
    except Exception:
        _FRAME_CACHE[key] = {}
        return {}


def _sum_frames(concepts: list[str], unit: str, periods: list[str]) -> dict[int, float]:
    ttm: dict[int, dict] = {}
    for concept in concepts:
        for period in periods:
            for cik, val in _fetch_one_frame(concept, unit, period).items():
                ttm.setdefault(cik, {})
                if period not in ttm[cik]:
                    ttm[cik][period] = val
    return {cik: sum(pv.values()) for cik, pv in ttm.items()}


def _best_frame(concepts: list[str], unit: str, periods: list[str]) -> dict[int, float]:
    best: dict[int, float] = {}
    for concept in concepts:
        for period in periods:
            for cik, val in _fetch_one_frame(concept, unit, period).items():
                if cik not in best:
                    best[cik] = val
    return best


def _build_cik_maps() -> tuple[dict[str, int], dict[int, str]]:
    try:
        resp = requests.get(
            "https://www.sec.gov/files/company_tickers.json",
            headers=_EDGAR_UA, timeout=20,
        )
        resp.raise_for_status()
        data = resp.json()
        ticker_to_cik = {}
        cik_to_ticker = {}
        for entry in data.values():
            t   = entry.get("ticker", "").upper().replace(".", "-")
            cik = int(entry["cik_str"])
            ticker_to_cik[t]   = cik
            cik_to_ticker[cik] = t
        return ticker_to_cik, cik_to_ticker
    except Exception as e:
        print(f"  [WARN] Could not build CIK maps: {e}")
        return {}, {}


def _compute_fundamentals(
    price: float,
    rev: float | None,    rev_prev: float | None,
    ni: float | None,     ni_prev: float | None,
    op_i: float | None,   d_a: float | None,
    ocf: float | None,    cx: float | None,
    tot_a: float | None,  eq: float | None,
    ca: float | None,     cl: float | None,
    csh: float | None,    ltd: float | None,
    sh: float | None,
) -> dict:
    """Derive all fundamental metrics from raw statement values."""
    mkt_cap   = (price * sh)             if sh and sh > 0             else None
    eps_trail = (ni / sh)                if ni is not None and sh and sh > 0 else None
    pe_trail  = (price / eps_trail)      if eps_trail and eps_trail > 0      else None
    pb        = (price / (eq / sh))      if eq and sh and sh > 0 and eq > 0  else None
    ebitda    = ((op_i + d_a)            if op_i is not None and d_a is not None else op_i)
    ev        = ((mkt_cap + (ltd or 0) - (csh or 0)) if mkt_cap is not None   else None)
    ev_ebitda = (ev / ebitda)            if ev and ebitda and ebitda > 0      else None
    ev_revenue= (ev / rev)               if ev and rev and rev > 0            else None
    roe       = (ni / eq)                if ni is not None and eq and eq > 0  else None
    roa       = (ni / tot_a)             if ni is not None and tot_a and tot_a > 0 else None
    de        = (ltd / eq)               if ltd is not None and eq and eq > 0 else None
    curr_r    = (ca / cl)                if ca and cl and cl > 0              else None
    pm        = (ni / rev)               if ni is not None and rev and rev > 0 else None
    om        = (op_i / rev)             if op_i is not None and rev and rev > 0 else None
    fcf       = ((ocf - cx)              if ocf is not None and cx is not None else ocf)
    fcf_yield = (fcf / mkt_cap)          if fcf is not None and mkt_cap and mkt_cap > 0 else None
    rev_g     = ((rev - rev_prev) / abs(rev_prev)
                 if rev is not None and rev_prev and rev_prev != 0 else None)
    ni_g      = ((ni - ni_prev) / abs(ni_prev)
                 if ni is not None and ni_prev and ni_prev != 0 else None)
    return {
        "market_cap":          mkt_cap,
        "shares_outstanding":  sh,
        "pe_trailing":         pe_trail,
        "pb_ratio":            pb,
        "ev_ebitda":           ev_ebitda,
        "ev_revenue":          ev_revenue,
        "eps_trailing":        eps_trail,
        "revenue":             rev,
        "revenue_growth":      rev_g,
        "earnings_growth":     ni_g,
        "roe":                 roe,
        "roa":                 roa,
        "debt_to_equity":      de,
        "total_debt":          ltd,
        "total_cash":          csh,
        "book_value":          (eq / sh) if eq and sh and sh > 0 else None,
        "current_ratio":       curr_r,
        "profit_margin":       pm,
        "operating_margin":    om,
        "fcf_yield":           fcf_yield,
    }


# ---------------------------------------------------------------------------
# Phase 3 — EDGAR bulk fundamentals (domestic quarterly filers)
# ---------------------------------------------------------------------------

def run_fundamentals_phase(all_tickers: list[str]) -> dict[str, dict]:
    """
    Fetch EDGAR XBRL frames for all domestic quarterly filers and return
    {ticker: fundamentals_dict}.  Foreign/annual filers handled separately.
    """
    print("Phase 3: Fetching EDGAR bulk fundamentals ...")
    ticker_to_cik, cik_to_ticker = _build_cik_maps()
    if not ticker_to_cik:
        print("  [WARN] CIK map unavailable — EDGAR bulk skipped")
        return {}

    quarters     = _get_recent_quarters(9)
    ttm_periods  = [f"CY{y}Q{q}"  for y, q in quarters[:4]]
    prev_periods = [f"CY{y}Q{q}"  for y, q in quarters[4:8]]
    bs_periods   = [f"CY{y}Q{q}I" for y, q in quarters[:3]]

    print(f"  TTM periods : {ttm_periods}")

    assets     = _best_frame(["Assets"],                                                           "USD",    bs_periods)
    equity     = _best_frame(["StockholdersEquity",
                               "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest",
                               "CommonStockholdersEquity"],                                         "USD", bs_periods)
    cur_assets = _best_frame(["AssetsCurrent"],                                                    "USD",    bs_periods)
    cur_liab   = _best_frame(["LiabilitiesCurrent"],                                               "USD",    bs_periods)
    cash       = _best_frame(["CashAndCashEquivalentsAtCarryingValue",
                               "CashCashEquivalentsAndShortTermInvestments"],                      "USD",    bs_periods)
    lt_debt    = _best_frame(["LongTermDebt", "LongTermDebtNoncurrent",
                               "LongTermDebtAndCapitalLeaseObligations",
                               "LongTermDebtAndFinanceLeaseLiabilities",
                               "FinanceLeaseLiabilityNoncurrent"],                                 "USD",    bs_periods)
    shares     = _best_frame(["CommonStockSharesOutstanding"],                                     "shares", bs_periods)
    revenue    = _sum_frames(["Revenues",
                               "RevenueFromContractWithCustomerExcludingAssessedTax",
                               "SalesRevenueNet"],                                                 "USD", ttm_periods)
    net_income = _sum_frames(["NetIncomeLoss"],                                                    "USD", ttm_periods)
    op_income  = _sum_frames(["OperatingIncomeLoss"],                                              "USD", ttm_periods)
    dna        = _sum_frames(["DepreciationDepletionAndAmortization",
                               "DepreciationAndAmortization"],                                     "USD", ttm_periods)
    op_cf      = _sum_frames(["NetCashProvidedByUsedInOperatingActivities"],                       "USD", ttm_periods)
    capex      = _sum_frames(["PaymentsToAcquirePropertyPlantAndEquipment"],                       "USD", ttm_periods)
    rev_prev   = _sum_frames(["Revenues",
                               "RevenueFromContractWithCustomerExcludingAssessedTax",
                               "SalesRevenueNet"],                                                 "USD", prev_periods)
    ni_prev    = _sum_frames(["NetIncomeLoss"],                                                    "USD", prev_periods)

    print(f"  Revenue coverage: {len(revenue)} companies")

    # Known foreign filer tickers — skip them here, handled in Phase 3B
    foreign_tickers = set(US_GAAP_ANNUAL_FILERS) | set(IFRS_ANNUAL_FILERS)

    result: dict[str, dict] = {}
    for ticker in all_tickers:
        if ticker in foreign_tickers:
            continue
        cik = ticker_to_cik.get(ticker)
        if cik is None:
            continue

        # yfinance price needed for ratios — use a fast single-ticker download
        try:
            raw = yf.download(ticker, period="5d", auto_adjust=True, progress=False)
            price = float(raw["Close"].dropna().iloc[-1]) if not raw.empty else None
        except Exception:
            price = None
        if price is None:
            continue

        metrics = _compute_fundamentals(
            price,
            revenue.get(cik),   rev_prev.get(cik),
            net_income.get(cik), ni_prev.get(cik),
            op_income.get(cik),  dna.get(cik),
            op_cf.get(cik),      capex.get(cik),
            assets.get(cik),     equity.get(cik),
            cur_assets.get(cik), cur_liab.get(cik),
            cash.get(cik),       lt_debt.get(cik),
            shares.get(cik),
        )
        metrics["ticker"]      = ticker
        metrics["filer_type"]  = "us-gaap-quarterly"
        metrics["updated_at"]  = datetime.now(timezone.utc).isoformat()
        result[ticker] = metrics

    print(f"  Domestic filers matched: {len(result)}\n")
    return result


# ---------------------------------------------------------------------------
# Phase 3B — Foreign / annual filer fundamentals (companyfacts per CIK)
# ---------------------------------------------------------------------------

def _fetch_companyfacts(cik_padded: str) -> dict:
    """Fetch all XBRL facts for one company from SEC companyfacts API."""
    url = f"https://data.sec.gov/api/xbrl/companyfacts/CIK{cik_padded}.json"
    try:
        resp = requests.get(url, headers=_EDGAR_UA, timeout=30)
        if resp.status_code == 404:
            return {}
        resp.raise_for_status()
        return resp.json().get("facts", {})
    except Exception:
        return {}


def _get_latest_annual_value(facts: dict, taxonomy: str, concept: str) -> float | None:
    """Extract the most recent annual (12-month) value for a concept."""
    try:
        units = facts.get(taxonomy, {}).get(concept, {}).get("units", {})
        # Try USD first, then shares, then any available unit
        for unit_key in ("USD", "shares", *units.keys()):
            entries = units.get(unit_key, [])
            # Filter to annual (form 10-K or 20-F) or 12-month duration entries
            annual = [
                e for e in entries
                if e.get("form") in ("10-K", "20-F")
                or (e.get("start") and e.get("end") and
                    _months_between(e["start"], e["end"]) >= 11)
            ]
            if not annual:
                continue
            # Pick most recent by end date
            annual.sort(key=lambda e: e.get("end", ""), reverse=True)
            return float(annual[0]["val"])
    except Exception:
        pass
    return None


def _get_prior_annual_value(facts: dict, taxonomy: str, concept: str) -> float | None:
    """Extract the second-most-recent annual value for YoY growth calculation."""
    try:
        units = facts.get(taxonomy, {}).get(concept, {}).get("units", {})
        for unit_key in ("USD", "shares", *units.keys()):
            entries = units.get(unit_key, [])
            annual = [
                e for e in entries
                if e.get("form") in ("10-K", "20-F")
                or (e.get("start") and e.get("end") and
                    _months_between(e["start"], e["end"]) >= 11)
            ]
            if len(annual) < 2:
                continue
            annual.sort(key=lambda e: e.get("end", ""), reverse=True)
            return float(annual[1]["val"])
    except Exception:
        pass
    return None


def _months_between(start: str, end: str) -> int:
    try:
        s = datetime.fromisoformat(start)
        e = datetime.fromisoformat(end)
        return (e.year - s.year) * 12 + (e.month - s.month)
    except Exception:
        return 0


def _extract_facts(facts: dict, taxonomy: str, concept_map: dict) -> dict:
    """Extract latest and prior annual values for all mapped concepts."""
    out = {}
    for concept, field in concept_map.items():
        val = _get_latest_annual_value(facts, taxonomy, concept)
        if val is not None and field not in out:
            out[field] = val
        prior = _get_prior_annual_value(facts, taxonomy, concept)
        if prior is not None and f"{field}_prev" not in out:
            out[f"{field}_prev"] = prior
    return out


def run_foreign_filers_phase() -> dict[str, dict]:
    """
    Fetch fundamentals for all known foreign/annual filers via the SEC
    companyfacts API.  Returns {ticker: fundamentals_dict}.
    """
    print("Phase 3B: Fetching foreign/annual filer fundamentals ...")
    result: dict[str, dict] = {}
    all_foreign = {
        **{t: (cik, "us-gaap", USGAAP_CONCEPT_MAP)  for t, cik in US_GAAP_ANNUAL_FILERS.items()},
        **{t: (cik, "ifrs-full", IFRS_CONCEPT_MAP)   for t, cik in IFRS_ANNUAL_FILERS.items()},
    }

    for ticker, (cik_padded, taxonomy, concept_map) in all_foreign.items():
        try:
            facts = _fetch_companyfacts(cik_padded)
            if not facts:
                print(f"  [WARN] No XBRL facts for {ticker} ({cik_padded})")
                continue

            raw = _extract_facts(facts, taxonomy, concept_map)

            # Get current price
            try:
                dl = yf.download(ticker, period="5d", auto_adjust=True, progress=False)
                price = float(dl["Close"].dropna().iloc[-1]) if not dl.empty else None
            except Exception:
                price = None
            if price is None:
                print(f"  [WARN] No price for {ticker} — skipping")
                continue

            metrics = _compute_fundamentals(
                price,
                raw.get("revenue"),        raw.get("revenue_prev"),
                raw.get("net_income"),     raw.get("net_income_prev"),
                raw.get("operating_income"), raw.get("dna"),
                raw.get("op_cf"),          raw.get("capex"),
                raw.get("assets"),         raw.get("equity"),
                raw.get("cur_assets"),     raw.get("cur_liabilities"),
                raw.get("cash"),           raw.get("lt_debt"),
                raw.get("shares"),
            )
            metrics["ticker"]     = ticker
            metrics["filer_type"] = f"{taxonomy}-annual"
            metrics["updated_at"] = datetime.now(timezone.utc).isoformat()
            result[ticker] = metrics
            print(f"  {ticker} ({taxonomy}) ✓")
            time.sleep(0.2)   # respect SEC rate limit
        except Exception as e:
            print(f"  [WARN] {ticker}: {e}")

    print(f"  Foreign filers done: {len(result)}\n")
    return result


# ---------------------------------------------------------------------------
# Phase 4 — FINRA short interest
# ---------------------------------------------------------------------------

def run_finra_phase(existing: dict[str, dict]) -> dict[str, dict]:
    """
    Fetch today's (or most recent) FINRA RegSHO short-volume file and merge
    short_percent into the fundamentals dict keyed by ticker.
    Returns the updated dict.
    """
    print("Phase 4: Fetching FINRA short interest ...")
    short_map: dict[str, float] = {}

    for days_back in range(1, 6):
        d = date.today() - timedelta(days=days_back)
        if d.weekday() >= 5:
            continue
        url = (
            "https://cdn.finra.org/equity/regsho/daily/"
            f"CNMSshvol{d.strftime('%Y%m%d')}.txt"
        )
        try:
            resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
            if resp.status_code == 404:
                continue
            resp.raise_for_status()
            for line in resp.text.splitlines()[1:]:
                parts = line.split("|")
                if len(parts) < 5:
                    continue
                ticker    = parts[0].strip()
                try:
                    short_vol = float(parts[1])
                    total_vol = float(parts[3])
                    if total_vol > 0:
                        short_map[ticker] = short_vol / total_vol
                except (ValueError, IndexError):
                    continue
            print(f"  FINRA short interest: {len(short_map)} tickers ({d})")
            break
        except Exception as e:
            print(f"  [WARN] FINRA {d}: {e}")
            continue

    if not short_map:
        print("  [WARN] FINRA short interest unavailable")

    # Merge into existing fundamentals dict
    now = datetime.now(timezone.utc).isoformat()
    for ticker, ratio in short_map.items():
        if ticker in existing:
            existing[ticker]["short_percent"] = ratio
        else:
            existing[ticker] = {
                "ticker":        ticker,
                "short_percent": ratio,
                "filer_type":    "finra-only",
                "updated_at":    now,
            }

    return existing


# ---------------------------------------------------------------------------
# Phase 5 — Sector statistics (median + MAD per metric per sector)
# ---------------------------------------------------------------------------

def _upsert_sector_stats(rows: list[dict]) -> None:
    """Upsert sector statistics rows into the sector_stats table."""
    _pg_upsert(f"{_TABLE_PREFIX}sector_stats", rows, conflict_col="sector")


def _load_sector_mapping() -> dict[str, str]:
    """
    Load ticker → sector mapping from the analyst_cache table that was just
    populated in Phase 1.  Returns {ticker: {sector, analyst_upside, eps_forward}}.
    """
    try:
        conn = _get_conn()
        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            cur.execute(
                f"SELECT ticker, sector, analyst_upside, eps_forward "
                f"FROM \"{_TABLE_PREFIX}analyst_cache\" WHERE sector IS NOT NULL"
            )
            rows = cur.fetchall()
        conn.close()
        return {r["ticker"]: dict(r) for r in rows}
    except Exception as e:
        print(f"  [WARN] Could not load sector mapping: {e}")
        return {}


def run_sector_stats_phase(fundamentals: dict[str, dict]) -> None:
    """
    Compute sector-level median and MAD for every fundamental metric and upsert
    to the sector_stats table.  Requires >= 15 tickers per sector.
    """
    import statistics as _stats

    print("Phase 5: Computing sector statistics ...")

    analyst_map = _load_sector_mapping()
    if not analyst_map:
        print("  [WARN] No analyst/sector data — sector stats skipped\n")
        return

    METRICS = [
        "pe_trailing", "pb_ratio", "ev_ebitda", "ev_revenue",
        "revenue_growth", "earnings_growth", "eps_growth",
        "roe", "roa", "debt_to_equity", "current_ratio",
        "profit_margin", "operating_margin", "fcf_yield",
        "analyst_upside",
    ]
    MIN_TICKERS = 15

    # Build merged per-ticker dict: fundamentals + analyst fields
    merged: dict[str, dict] = {}
    for ticker, fdata in fundamentals.items():
        adata = analyst_map.get(ticker, {})
        sector = adata.get("sector")
        if not sector or sector == "N/A":
            continue
        eps_t = fdata.get("eps_trailing")
        eps_f = adata.get("eps_forward")
        eps_growth = None
        if eps_t and eps_f and abs(eps_t) > 1e-9:
            raw_g = (eps_f - eps_t) / abs(eps_t)
            # Clamp to ±150% to avoid breakeven-crossing distortion
            eps_growth = max(-1.5, min(1.5, raw_g))
        merged[ticker] = {
            **{m: fdata.get(m) for m in METRICS},
            "eps_growth":    eps_growth,
            "analyst_upside": adata.get("analyst_upside"),
            "_sector":       sector,
        }

    # Group by sector
    sector_buckets: dict[str, list[dict]] = {}
    for data in merged.values():
        s = data["_sector"]
        sector_buckets.setdefault(s, []).append(data)

    now = datetime.now(timezone.utc).isoformat()
    rows = []
    for sector, members in sector_buckets.items():
        if len(members) < MIN_TICKERS:
            continue
        row: dict = {"sector": sector, "ticker_count": len(members), "updated_at": now}
        for metric in METRICS:
            vals = [m[metric] for m in members
                    if m.get(metric) is not None
                    and isinstance(m[metric], (int, float))
                    and m[metric] == m[metric]]  # exclude NaN
            if len(vals) >= MIN_TICKERS:
                med = _stats.median(vals)
                mad = _stats.median([abs(v - med) for v in vals])
                row[f"{metric}_med"] = med
                row[f"{metric}_mad"] = max(float(mad), 1e-10)
            else:
                row[f"{metric}_med"] = None
                row[f"{metric}_mad"] = None
        rows.append(row)

    if rows:
        _upsert_sector_stats(rows)
        print(f"  Sector stats written for {len(rows)} sectors")
    else:
        print("  [WARN] No sectors met the minimum ticker threshold")

    print(f"  Phase 5 complete\n")


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main() -> None:
    print("=" * 55)
    print("  ANALYST + NEWS + FUNDAMENTALS CACHE BUILDER")
    print("=" * 55 + "\n")

    tickers = fetch_tickers()
    if not tickers:
        print("ERROR: No tickers fetched — aborting.")
        raise SystemExit(1)

    # Phase 1 — analyst consensus + forward estimates (yfinance)
    run_analyst_phase(tickers)

    # Phase 2 — news sentiment (Google News RSS + VADER)
    run_news_phase(tickers)

    # Phase 3 — EDGAR bulk fundamentals (domestic quarterly filers)
    fundamentals = run_fundamentals_phase(tickers)

    # Phase 3B — foreign / annual filer fundamentals (SEC companyfacts)
    foreign = run_foreign_filers_phase()
    fundamentals.update(foreign)

    # Phase 4 — FINRA short interest (merged into fundamentals)
    fundamentals = run_finra_phase(fundamentals)

    # Write all fundamentals to Supabase
    print(f"Writing {len(fundamentals)} rows to fundamentals_cache ...")
    batch: list[dict] = []
    for row in fundamentals.values():
        batch.append(row)
        if len(batch) >= UPSERT_EVERY:
            _upsert_fundamentals(batch)
            batch.clear()
    if batch:
        _upsert_fundamentals(batch)

    # Phase 5 — Sector statistics (median + MAD per metric per sector)
    run_sector_stats_phase(fundamentals)

    print("\nCache build complete.")


def phase5_only() -> None:
    """Load fundamentals from DB and recompute sector stats — no cache rebuild."""
    print("=" * 55)
    print("  PHASE 5 ONLY — Sector statistics recompute")
    print(f"  TABLE_PREFIX={_TABLE_PREFIX!r}")
    print("=" * 55 + "\n")

    print(f"Loading fundamentals from {_TABLE_PREFIX}fundamentals_cache ...")
    cols = [c for c in _FUND_COLS if c != "updated_at"]
    col_str = ", ".join(f'"{c}"' for c in cols)
    try:
        conn = _get_conn()
        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
            cur.execute(f'SELECT {col_str} FROM "{_TABLE_PREFIX}fundamentals_cache"')
            rows = cur.fetchall()
        conn.close()
    except Exception as e:
        print(f"ERROR: Could not load fundamentals: {e}")
        raise SystemExit(1)

    fundamentals = {r["ticker"]: dict(r) for r in rows}
    print(f"  Loaded {len(fundamentals)} rows\n")

    run_sector_stats_phase(fundamentals)
    print("\nPhase 5 complete.")


if __name__ == "__main__":
    import sys
    if "--phase5-only" in sys.argv:
        phase5_only()
    else:
        main()