ssz223
/
Stock-Tool
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Stock-Tool/cache_builder.py

1201 lines
48 KiB
Python
Raw Blame History

"""
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")
if not _DB_PASS:
raise RuntimeError("DB_PASS environment variable is not set")
# 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 = 15.0 # slower main pass — reduces Yahoo throttling
MAX_WORKERS = 2 # fewer concurrent workers
NEWS_WORKERS = 10 # reduced from 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
# A throttled Yahoo response returns a sparse dict (< ~30 keys).
# Reject it so the ticker is queued for the retry pass instead of
# being stored with all-NULL analyst fields.
if not info or not isinstance(info, dict) or len(info) < 30:
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,
"name": info.get("longName") or None,
"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,
"short_percent": info.get("shortPercentOfFloat"),
"short_ratio": info.get("shortRatio"),
"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 main pass complete in {time.time() - start:.0f}s\n")
_retry_analyst_pass(tickers)
def _retry_analyst_pass(tickers: list[str]) -> None:
"""
Single-threaded retry for tickers whose Phase 1 fetch returned a sparse
(throttled) response. Queries DB for rows with NULL analyst_count and
re-fetches them one at a time with a 30-second delay between requests.
"""
try:
conn = _get_conn()
with conn.cursor() as cur:
cur.execute(
f'SELECT ticker FROM "{_TABLE_PREFIX}analyst_cache" '
f'WHERE analyst_count IS NULL'
)
nulls = {r[0] for r in cur.fetchall()}
conn.close()
except Exception as e:
print(f" [WARN] Retry query failed: {e}\n")
return
retry = [t for t in tickers if t in nulls]
if not retry:
print(" Retry pass: all tickers have analyst data.\n")
return
print(f" Retry pass: {len(retry)} tickers need re-fetch (30s delay each) ...")
pending: list[dict] = []
recovered = 0
for i, ticker in enumerate(retry, 1):
print(f"\r {i}/{len(retry)} {ticker} ", end="", flush=True)
result = _fetch_analyst(ticker)
if result and result.get("analyst_count") is not None:
pending.append(result)
recovered += 1
if len(pending) >= UPSERT_EVERY:
print()
_upsert(pending)
pending.clear()
time.sleep(30.0)
if pending:
print()
_upsert(pending)
print(f"\n Retry pass complete — recovered {recovered}/{len(retry)} tickers\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,
"news_headline_count": len(scores),
"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", "name", "pe_forward", "eps_forward", "analyst_norm", "analyst_upside",
"analyst_count", "analyst_target", "recommendation", "news_sentiment",
"news_headline_count", "sector", "industry", "short_percent", "short_ratio", "updated_at",
)
_FUND_COLS = (
"ticker", "name", "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",
"piotroski_score", "accruals_ratio",
"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 requests.HTTPError as e:
print(f" [WARN] EDGAR frame {concept}/{unit}/{period}: HTTP {e.response.status_code} — not cached")
return {}
except Exception as e:
print(f" [WARN] EDGAR frame {concept}/{unit}/{period}: {e} — not cached")
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() if len(pv) >= 4}
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)
# Piotroski partial F-Score (4 signals available at compute time)
p = 0
if ni is not None and tot_a and tot_a > 0 and ni / tot_a > 0: p += 1 # ROA > 0
if ocf is not None and ocf > 0: p += 1 # OCF > 0
if ocf is not None and ni is not None and ocf > ni: p += 1 # OCF > NI (cash-backed)
if rev is not None and rev_prev is not None and rev_prev != 0 and rev > rev_prev: p += 1 # Revenue improving
piotroski = p
# Sloan accruals ratio: (NI - OCF) / Total Assets — negative = higher earnings quality
accruals = None
if ni is not None and ocf is not None and tot_a and tot_a > 0:
accruals = max(-0.3, min(0.3, (ni - ocf) / tot_a))
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,
"piotroski_score": piotroski,
"accruals_ratio": accruals,
}
# ---------------------------------------------------------------------------
# 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] = {}
price_failed: list[str] = []
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:
price_failed.append(ticker)
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),
)
try:
_info = yf.Ticker(ticker).info or {}
_name = _info.get("longName") or _info.get("shortName") or None
except Exception:
_name = None
metrics["ticker"] = ticker
metrics["name"] = _name
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")
# Retry tickers whose price fetch was empty (likely throttled).
if price_failed:
print(f" Price retry: {len(price_failed)} tickers (10s delay each) ...")
retried = 0
for i, ticker in enumerate(price_failed, 1):
print(f"\r {i}/{len(price_failed)} {ticker} ", end="", flush=True)
cik = ticker_to_cik.get(ticker)
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 not None and cik is not None:
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),
)
try:
_info = yf.Ticker(ticker).info or {}
_name = _info.get("longName") or _info.get("shortName") or None
except Exception:
_name = None
metrics["ticker"] = ticker
metrics["name"] = _name
metrics["filer_type"] = "us-gaap-quarterly"
metrics["updated_at"] = datetime.now(timezone.utc).isoformat()
result[ticker] = metrics
retried += 1
time.sleep(10.0)
print(f"\n Price retry done — recovered {retried}/{len(price_failed)} tickers\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"),
)
try:
_info = yf.Ticker(ticker).info or {}
_name = _info.get("longName") or _info.get("shortName") or None
except Exception:
_name = None
metrics["ticker"] = ticker
metrics["name"] = _name
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()
Reference in New Issue View Git Blame Copy Permalink