630 lines
25 KiB
Python
630 lines
25 KiB
Python
import argparse
|
||
import sqlite3, sys, time, re, requests, random
|
||
from typing import List, Tuple, Set, Optional, Iterable
|
||
import pandas as pd
|
||
import yfinance as yf
|
||
from bs4 import BeautifulSoup
|
||
from datetime import datetime, timedelta, timezone
|
||
from io import StringIO
|
||
try:
|
||
from zoneinfo import ZoneInfo
|
||
CT = ZoneInfo("America/Chicago")
|
||
except Exception:
|
||
CT = None
|
||
|
||
|
||
|
||
# ========= USER CONFIG =======================================================
|
||
DB_PATH = r"C:\Users\srk47\Desktop\options_flow.db"
|
||
SEGMENTS = ["large", "mid", "nasdaq"] # auto universes to include
|
||
TAG_AUTO_SEGMENTS = True # tag auto symbols as AUTO:SP500/SP400/NDX100
|
||
EXTRA_SYMBOLS = [] # optional flat list, e.g. ["TSLA","COIN","SPY"]
|
||
|
||
EXTRA_GROUPS = {
|
||
"OIL": ["HUSA","INDO","USEG","MARPS","BRN","PED","GBR","KLXE","CEIN","IMPP","CVX","OXY","GUSH","RCON","APA","VIVK"],
|
||
"FASTFOOD_REST": ["SG","CMG","WEN","QSR","WING","BROS","SHAK","CAVA","SBUX","MCD","YUM"],
|
||
"ROBOTICS": ["SERV","RR","KITT","MBOT","IRBT","ARBE","SYM","TSLA","NVDA","ROBO","BOTZ","TER","ROK"],
|
||
"SEMIS": ["NVDA","MCHP","NVTS","ADI","QCOM","AVGO","ON","TXN","QQQ","OLED","STM","ASML","RMBS","ARM","SMH","INTC","LRCX","DELL","MRVL","TSM","SMCI","MU","SOXL","AMD"],
|
||
"QUANTUM": ["IONQ","RGTI","QUBT","QBTS","ARQQ","QMCO","COHR","MRVL","HON","TSEM","FORM"],
|
||
"CYBER": ["CYBR","FTNT","PANW","ZS","CRWD","OKTA","S"],
|
||
"FINTECH": ["SOFI","AFRM","NU","V","MA","PYPL","UPST","FOUR","AXP","PSFE","PAYO","ADYEY","QFIN"],
|
||
"CATHY": ["TXG","TWST","CRSP","COIN","PACB","DNA","SOFI","PD","HOOD","ROKU","ZM","EXAS","U","NTLA","SPY","TER","PATH","PINS","META","ARKK","CERS","BEAM","VCYT","RBLX","TTD","TSLA","TDOC","ACHR"],
|
||
"CHINA": ["FUTU","KC","LI","EDU","BILI","KWEB","PDD","JD","BABA","BIDU","NIO","XPEV","GDS"],
|
||
"SOLAR": ["DQ","SEDG","JKS","ENPH","TAN","CSIQ","ARRY","FSLR","RUN"],
|
||
"BANKS": ["MS","C","GS","KRE","WFC","JPM","BAC","SCHW","XLF"],
|
||
"TRAVEL": ["EXPE","BKNG","WYNN","LVS","TRIP","ABNB","JETS","RCL","LUV","DAL","UAL","NCLH","CCL","MAR","HLT"],
|
||
"WEED": ["ACB","MSOS","CGC","CURLF","GTBIF","TLRY","SNDL","MSOX","MJ"],
|
||
"DOW": ["WMT","CRM","PG","HD","MSFT","V","MCD","IBM","AMGN","VZ","CSCO","AAPL","MRK","DIS","TRV","HON","JNJ","JPM","NKE","AXP","MMM","GS","CVX","CAT","UNH","BA","BE"],
|
||
"NUCLEAR": ["SMR","OKLO","NNE","ASPI","LTBR","LEU","TLN","CEG","PEG"],
|
||
"URANIUM": ["URA","NLR","URNM","NXE","URAN","URNJ","CCJ","UEC"],
|
||
"SAAS-CLOUD": ["NOW","CRM","WDAY","OKTA","MDB","DDOG","TWLO","DOCU","WIX","TEAM","SNOW","NET","U","PLTR","BILL","DOCN","VEEV","CRWV"],
|
||
"SPACE-ROCKET": ["RKLB","SPCE","SIDU","MNTS","RDW","LUNR","ASTS","SATS","ARKX","VSAT","IRDM","SES"],
|
||
"MEME": ["NEGG","AMC","BBAI","BYND","OPEN","UPST","NVAX","GME","DJT","BB","TRUP","KOSS","KODK","SPWR","RDDT","HOOD"],
|
||
"DRONES": ["NOC","LMT","KTOS","BA","AIRO","UMAC","DPRO","ONDS","RTX","RCAT","EVTL","AMBA","ACHR","JOBY","GPRO","HON","GOGO","EH"]
|
||
}
|
||
SYMBOLS_ONLY = False # True = ignore SEGMENTS and use only EXTRA_SYMBOLS + EXTRA_GROUPS
|
||
LOOKBACK_DAYS_INTRADAY = 1 # 1–7 for 1m
|
||
DAILY_PERIOD = "2y"
|
||
BATCH_SIZE = 110 # 50–120 good
|
||
SLEEP_BETWEEN_BATCHES = 0.25 # seconds
|
||
LOOP = True
|
||
INTRADAY_INTERVAL_MIN = 1
|
||
DAILY_INTERVAL_MIN = 1
|
||
FORCE_FIX_CST = True # drop legacy *_cst tables (once)
|
||
# ============================================================================
|
||
|
||
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
|
||
HEADERS = {"User-Agent": USER_AGENT}
|
||
URL_SP500 = "https://en.wikipedia.org/wiki/List_of_S%26P_500_companies"
|
||
URL_SP400 = "https://en.wikipedia.org/wiki/List_of_S%26P_400_companies"
|
||
URL_NASDAQ = "https://en.wikipedia.org/wiki/Nasdaq-100"
|
||
VALID_TICKER_RE = re.compile(r"^[A-Z][A-Z0-9.\-]{0,6}$")
|
||
BAD_WORDS = {"CLOSING","INTRADAY","TICKER","SYMBOL","INDEX","COMPANY","WEIGHT","WEIGHTS"}
|
||
|
||
|
||
# ---------- time helpers ----------
|
||
def now_utc() -> str:
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
|
||
def now_ct() -> str:
|
||
if CT:
|
||
return datetime.now(CT).strftime("%Y-%m-%d %I:%M:%S %p CT")
|
||
return time.strftime("%Y-%m-%d %I:%M:%S %p (local)")
|
||
def stamp(label: str) -> str:
|
||
return f"[{label}] {now_ct()} | {now_utc()}"
|
||
|
||
|
||
# ---------- db ----------
|
||
def connect_rw(db_path: str) -> sqlite3.Connection:
|
||
con = sqlite3.connect(db_path, timeout=10.0, isolation_level=None)
|
||
con.execute("PRAGMA journal_mode=WAL;")
|
||
con.execute("PRAGMA synchronous=NORMAL;")
|
||
con.execute("PRAGMA busy_timeout=5000;")
|
||
con.execute("PRAGMA temp_store=MEMORY;")
|
||
con.execute("PRAGMA mmap_size=268435456;")
|
||
con.execute("PRAGMA cache_size=-80000;")
|
||
return con
|
||
|
||
|
||
def ensure_metadata_tables(con: sqlite3.Connection):
|
||
con.execute("""
|
||
CREATE TABLE IF NOT EXISTS universe_symbols (
|
||
symbol TEXT PRIMARY KEY,
|
||
source TEXT,
|
||
added_at TEXT
|
||
)""")
|
||
con.execute("""
|
||
CREATE TABLE IF NOT EXISTS symbol_sectors (
|
||
symbol TEXT NOT NULL,
|
||
sector TEXT NOT NULL,
|
||
PRIMARY KEY (symbol, sector)
|
||
)""")
|
||
|
||
def ensure_prices_daily_schema(con: sqlite3.Connection):
|
||
con.execute("""
|
||
CREATE TABLE IF NOT EXISTS prices_daily (
|
||
symbol TEXT NOT NULL,
|
||
date TEXT NOT NULL, -- 'YYYY-MM-DD'
|
||
open REAL, high REAL, low REAL, close REAL,
|
||
adj_close REAL, volume REAL,
|
||
source TEXT,
|
||
PRIMARY KEY (symbol, date)
|
||
)""")
|
||
con.execute("""
|
||
CREATE INDEX IF NOT EXISTS idx_prices_daily_symbol_date
|
||
ON prices_daily(symbol, date DESC)
|
||
""")
|
||
def ensure_intraday_schema(con: sqlite3.Connection):
|
||
con.execute("""
|
||
CREATE TABLE IF NOT EXISTS prices_intraday_1m (
|
||
symbol TEXT NOT NULL,
|
||
ts INTEGER NOT NULL, -- epoch seconds (UTC, minute-aligned)
|
||
open REAL,
|
||
high REAL,
|
||
low REAL,
|
||
close REAL,
|
||
volume REAL,
|
||
source TEXT,
|
||
date_cst TEXT, -- 'YYYY-MM-DD' (CT session date)
|
||
min_cst INTEGER, -- 0..1439 (CT minute-of-day)
|
||
PRIMARY KEY (symbol, ts)
|
||
)""")
|
||
con.execute("""
|
||
CREATE INDEX IF NOT EXISTS idx_prices_intraday_symbol_ts
|
||
ON prices_intraday_1m(symbol, ts DESC)
|
||
""")
|
||
con.execute("""
|
||
CREATE INDEX IF NOT EXISTS idx_p1m_sym_date_min
|
||
ON prices_intraday_1m(symbol, date_cst, min_cst)
|
||
""")
|
||
con.execute("""
|
||
CREATE INDEX IF NOT EXISTS idx_p1m_date_min_sym
|
||
ON prices_intraday_1m(date_cst, min_cst, symbol)
|
||
""")
|
||
|
||
|
||
def ensure_views(con: sqlite3.Connection, *, force_fix_cst: bool = False):
|
||
"""
|
||
Recreate *_cst views. If legacy tables exist with same names, drop them.
|
||
"""
|
||
def drop_any(name: str):
|
||
row = con.execute("SELECT type FROM sqlite_master WHERE name=?", (name,)).fetchone()
|
||
if not row: return
|
||
typ = row[0]
|
||
if typ == "view":
|
||
con.execute(f"DROP VIEW IF EXISTS {name}")
|
||
elif typ == "table":
|
||
con.execute(f"DROP TABLE IF EXISTS {name}")
|
||
else:
|
||
try: con.execute(f"DROP VIEW IF EXISTS {name}")
|
||
except Exception:
|
||
try: con.execute(f"DROP TABLE IF EXISTS {name}")
|
||
except Exception: pass
|
||
|
||
if force_fix_cst:
|
||
drop_any("prices_intraday_1m_cst")
|
||
drop_any("prices_daily_cst")
|
||
|
||
drop_any("prices_intraday_1m_cst")
|
||
drop_any("prices_daily_cst")
|
||
|
||
con.execute("""
|
||
CREATE VIEW IF NOT EXISTS prices_daily_cst AS
|
||
SELECT symbol, date AS date_cst,
|
||
open, high, low, close, adj_close, volume, source
|
||
FROM prices_daily
|
||
""")
|
||
con.execute("""
|
||
CREATE VIEW IF NOT EXISTS prices_intraday_1m_cst AS
|
||
SELECT
|
||
symbol,
|
||
datetime(ts,'unixepoch','localtime') AS dt_cst, -- wall clock
|
||
ts AS ts_cst, -- keep UTC epoch
|
||
open, high, low, close, volume, source
|
||
FROM prices_intraday_1m
|
||
""")
|
||
# Useful utility: latest intraday per symbol (UTC ts)
|
||
con.execute("""
|
||
CREATE VIEW IF NOT EXISTS v_latest_intraday_minute AS
|
||
SELECT symbol, MAX(ts) AS last_ts
|
||
FROM prices_intraday_1m
|
||
GROUP BY symbol
|
||
""")
|
||
|
||
def get_last_daily_date(con: sqlite3.Connection, sym: str) -> Optional[str]:
|
||
row = con.execute("SELECT MAX(date) FROM prices_daily WHERE symbol=?", (sym,)).fetchone()
|
||
return row[0] if row and row[0] else None
|
||
|
||
|
||
# ---------- symbol universe ----------
|
||
def normalize_yahoo_symbol(s: str) -> str:
|
||
return s.strip().upper().replace(".", "-")
|
||
|
||
def _unique(seq: Iterable[str]) -> List[str]:
|
||
seen, out = set(), []
|
||
for x in seq:
|
||
if x not in seen:
|
||
out.append(x); seen.add(x)
|
||
return out
|
||
|
||
def _parse_first_wikitable_with_columns(html: str, preferred_cols: Tuple[str, ...]) -> List[str]:
|
||
soup = BeautifulSoup(html, "html.parser")
|
||
for t in soup.select("table.wikitable"):
|
||
try:
|
||
df = pd.read_html(StringIO(str(t)))[0]
|
||
except Exception:
|
||
continue
|
||
for pc in preferred_cols:
|
||
if pc in df.columns:
|
||
ser = (df[pc].astype(str)
|
||
.str.replace(r"\[.*?\]", "", regex=True)
|
||
.str.strip())
|
||
out = []
|
||
for raw in ser.tolist():
|
||
up = raw.upper().replace(" ", "").replace("\u200b","")
|
||
if up in BAD_WORDS: # safety against header junk
|
||
continue
|
||
if VALID_TICKER_RE.match(up):
|
||
out.append(normalize_yahoo_symbol(up))
|
||
if out:
|
||
return out
|
||
return []
|
||
|
||
def build_ticker_universe(segments: List[str]) -> Tuple[List[str], dict]:
|
||
# manual group tags
|
||
sector_map = {}
|
||
manual_group_syms = []
|
||
for sector, lst in (EXTRA_GROUPS or {}).items():
|
||
for s in lst:
|
||
sym = normalize_yahoo_symbol(s)
|
||
manual_group_syms.append(sym)
|
||
sector_map.setdefault(sym, set()).add(sector)
|
||
manual_group_syms = _unique(manual_group_syms)
|
||
|
||
manual_flat = [normalize_yahoo_symbol(s) for s in EXTRA_SYMBOLS]
|
||
for s in manual_flat:
|
||
sector_map.setdefault(s, set()).add("MANUAL_LIST")
|
||
|
||
if SYMBOLS_ONLY:
|
||
return _unique(manual_group_syms + manual_flat), sector_map
|
||
|
||
auto_syms: List[str] = []
|
||
if "large" in segments:
|
||
print("Fetching S&P 500 tickers…")
|
||
r = requests.get(URL_SP500, headers=HEADERS, timeout=20); r.raise_for_status()
|
||
spx = _parse_first_wikitable_with_columns(r.text, ("Symbol","Ticker","Ticker symbol"))
|
||
auto_syms += spx
|
||
if TAG_AUTO_SEGMENTS:
|
||
for s in spx: sector_map.setdefault(s, set()).add("AUTO:SP500")
|
||
|
||
if "mid" in segments:
|
||
print("Fetching S&P 400 tickers…")
|
||
r = requests.get(URL_SP400, headers=HEADERS, timeout=20); r.raise_for_status()
|
||
mid = _parse_first_wikitable_with_columns(r.text, ("Symbol","Ticker","Ticker symbol"))
|
||
auto_syms += mid
|
||
if TAG_AUTO_SEGMENTS:
|
||
for s in mid: sector_map.setdefault(s, set()).add("AUTO:SP400")
|
||
|
||
if "nasdaq" in segments:
|
||
print("Fetching Nasdaq-100 tickers…")
|
||
r = requests.get(URL_NASDAQ, headers=HEADERS, timeout=20); r.raise_for_status()
|
||
ndx = _parse_first_wikitable_with_columns(r.text, ("Ticker","Ticker symbol","Symbol"))
|
||
auto_syms += ndx
|
||
if TAG_AUTO_SEGMENTS:
|
||
for s in ndx: sector_map.setdefault(s, set()).add("AUTO:NDX100")
|
||
|
||
return _unique(auto_syms + manual_group_syms + manual_flat), sector_map
|
||
|
||
|
||
# ---------- yfinance helpers ----------
|
||
def yf_download_multi(tickers: List[str], period: str, interval: str) -> pd.DataFrame:
|
||
return yf.download(tickers=tickers, period=period, interval=interval,
|
||
auto_adjust=False, progress=False, threads=True, group_by="ticker")
|
||
|
||
def chunks(seq, n):
|
||
for i in range(0, len(seq), n):
|
||
yield seq[i:i+n]
|
||
|
||
|
||
# ---------- metadata upserts ----------
|
||
def ensure_meta(con: sqlite3.Connection):
|
||
ensure_metadata_tables(con)
|
||
|
||
def upsert_universe(con: sqlite3.Connection, symbols: List[str], sector_map: dict):
|
||
now_iso = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
rows = []
|
||
for s in symbols:
|
||
tags = sector_map.get(s, set())
|
||
is_manual = ("MANUAL_LIST" in tags) or any(not t.startswith("AUTO:") for t in tags)
|
||
rows.append((s, "manual" if is_manual else "auto", now_iso))
|
||
con.executemany("""
|
||
INSERT OR IGNORE INTO universe_symbols(symbol, source, added_at)
|
||
VALUES(?,?,?)
|
||
""", rows)
|
||
|
||
pairs = []
|
||
for s, tags in sector_map.items():
|
||
for t in tags:
|
||
pairs.append((s, t))
|
||
if pairs:
|
||
con.executemany("""
|
||
INSERT OR IGNORE INTO symbol_sectors(symbol, sector)
|
||
VALUES(?,?)
|
||
""", pairs)
|
||
con.commit()
|
||
|
||
|
||
# ---------- data upserts ----------
|
||
def upsert_daily(con: sqlite3.Connection, sym: str, df: pd.DataFrame, source="yfinance", diag: bool=False) -> int:
|
||
if df is None or df.empty:
|
||
print(f"[daily upsert] {sym}: EMPTY dataframe from yfinance")
|
||
return 0
|
||
|
||
# Handle single or multi-ticker DataFrame
|
||
if isinstance(df.columns, pd.MultiIndex):
|
||
g = None
|
||
for level in (0, 1):
|
||
try:
|
||
g = df.xs(sym, axis=1, level=level)
|
||
break
|
||
except Exception:
|
||
pass
|
||
if g is None:
|
||
print(f"[daily upsert] {sym}: could not slice MultiIndex for ticker")
|
||
return 0
|
||
else:
|
||
g = df
|
||
|
||
g = g.copy()
|
||
idx = pd.to_datetime(g.index, errors="coerce")
|
||
g = g[~idx.isna()]
|
||
try:
|
||
idx = idx.tz_localize(None)
|
||
except Exception:
|
||
pass
|
||
g.index = idx
|
||
|
||
# Normalize column names to our schema
|
||
g = g.rename(columns={
|
||
"Open":"open", "High":"high", "Low":"low", "Close":"close",
|
||
"Adj Close":"adj_close", "Volume":"volume"
|
||
})
|
||
for c in ["open","high","low","close","adj_close","volume"]:
|
||
if c not in g.columns:
|
||
g[c] = None
|
||
|
||
g["date"] = g.index.strftime("%Y-%m-%d")
|
||
g["symbol"] = sym
|
||
g["source"] = source
|
||
|
||
if diag:
|
||
dmin = g["date"].min() if len(g) else "NA"
|
||
dmax = g["date"].max() if len(g) else "NA"
|
||
print(f"[daily upsert] {sym}: normalized rows={len(g)} range=({dmin} → {dmax})")
|
||
try:
|
||
g.reset_index(drop=True).to_csv(f"daily_{sym}.csv", index=False)
|
||
print(f"[diag] wrote preview CSV → daily_{sym}.csv")
|
||
except Exception as e:
|
||
print(f"[diag] CSV error: {e}")
|
||
|
||
rows = g[["symbol","date","open","high","low","close","adj_close","volume","source"]].values.tolist()
|
||
try:
|
||
con.executemany("""
|
||
INSERT INTO prices_daily(symbol,date,open,high,low,close,adj_close,volume,source)
|
||
VALUES (?,?,?,?,?,?,?,?,?)
|
||
ON CONFLICT(symbol,date) DO UPDATE SET
|
||
open=excluded.open, high=excluded.high, low=excluded.low, close=excluded.close,
|
||
adj_close=excluded.adj_close, volume=excluded.volume, source=excluded.source
|
||
""", rows)
|
||
except Exception as e:
|
||
print(f"[daily upsert] {sym}: sqlite error -> {e}")
|
||
return 0
|
||
return len(rows)
|
||
|
||
# ---------- single-symbol seeder (CLI) ----------
|
||
def seed_single_daily(con, sym: str, period: str = "2y", diag: bool=False) -> None:
|
||
print(stamp(f"SEED DAILY {sym} (period={period})"))
|
||
df = yf.download(sym, period=period, interval="1d", auto_adjust=False, progress=False)
|
||
if df is None or df.empty:
|
||
print(f"[SEED] {sym}: yfinance returned EMPTY — cannot seed")
|
||
return
|
||
|
||
idx = pd.to_datetime(df.index, errors="coerce")
|
||
try: idx = idx.tz_localize(None)
|
||
except Exception: pass
|
||
dmin = str(pd.to_datetime(idx.min()).date()) if len(idx) else "NA"
|
||
dmax = str(pd.to_datetime(idx.max()).date()) if len(idx) else "NA"
|
||
print(f"[SEED] {sym}: fetched {len(df)} rows ({dmin} → {dmax})")
|
||
|
||
wrote = upsert_daily(con, sym, df, diag=diag)
|
||
con.commit()
|
||
print(f"[SEED] {sym}: inserted/updated {wrote} rows")
|
||
|
||
row = con.execute("""
|
||
SELECT COUNT(*), MIN(date), MAX(date)
|
||
FROM prices_daily
|
||
WHERE symbol=?
|
||
""", (sym,)).fetchone()
|
||
print(f"[SEED CHECK] {sym}: count={row[0]} range=({row[1]} → {row[2]})")
|
||
|
||
|
||
def upsert_intraday(con: sqlite3.Connection, sym: str, df: pd.DataFrame, source="yfinance") -> int:
|
||
if df is None or df.empty:
|
||
return 0
|
||
|
||
# Slice the multi-index (group_by="ticker") or pass-through single frame
|
||
g = df.xs(sym, axis=1, level=0) if isinstance(df.columns, pd.MultiIndex) else df
|
||
if g is None or g.empty:
|
||
return 0
|
||
|
||
g = g.rename(columns={"Open": "open", "High": "high", "Low": "low", "Close": "close", "Volume": "volume"}).copy()
|
||
|
||
# Ensure UTC time index
|
||
idx = pd.to_datetime(g.index, errors="coerce")
|
||
if idx.tz is None:
|
||
idx = idx.tz_localize("UTC", nonexistent="shift_forward", ambiguous="NaT")
|
||
else:
|
||
idx = idx.tz_convert("UTC")
|
||
|
||
# Epoch seconds as pandas Series aligned to g
|
||
ts = pd.Series((idx.view("int64") // 10**9).astype("int64"), index=g.index)
|
||
|
||
# Central Time (DST-aware) session keys → plain NumPy arrays (no .values needed)
|
||
dt_ct = idx.tz_convert("America/Chicago")
|
||
date_cst = dt_ct.strftime("%Y-%m-%d") # this is a pandas Index[str] → array-like is fine
|
||
min_cst = (dt_ct.hour.astype(int) * 60 + dt_ct.minute.astype(int)).astype("int64") # NumPy array
|
||
|
||
# Populate columns (array-likes are OK as long as the lengths match)
|
||
g["symbol"] = sym.upper()
|
||
g["ts"] = ts # pandas Series
|
||
g["date_cst"] = date_cst # array-like
|
||
g["min_cst"] = min_cst # array-like
|
||
g["source"] = source
|
||
|
||
# Order matches table columns
|
||
rows = g[["symbol","ts","open","high","low","close","volume","source","date_cst","min_cst"]].itertuples(index=False, name=None)
|
||
|
||
# Use INSERT OR REPLACE for broad SQLite compatibility
|
||
with con:
|
||
con.executemany("""
|
||
INSERT OR REPLACE INTO prices_intraday_1m
|
||
(symbol, ts, open, high, low, close, volume, source, date_cst, min_cst)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?)
|
||
""", list(rows))
|
||
|
||
return len(g)
|
||
|
||
|
||
|
||
# ---------- runners ----------
|
||
def run_intraday_batched(con, symbols, lookback_days, batch_size, sleep_between):
|
||
period = f"{lookback_days}d"
|
||
total_rows, errors = 0, 0
|
||
print(stamp(f"INTRADAY START last {lookback_days}d"))
|
||
print(f"Universe size: {len(symbols)}; batch={batch_size}", flush=True)
|
||
|
||
for bi, batch in enumerate(chunks(symbols, batch_size), 1):
|
||
tag = f"[batch {bi}/{(len(symbols)+batch_size-1)//batch_size}]"
|
||
print(stamp(f"{tag} download {len(batch)} tickers"), flush=True)
|
||
try:
|
||
df = yf_download_multi(batch, period=period, interval="1m")
|
||
except KeyboardInterrupt:
|
||
print("\n[interrupt] stopping intraday loop gracefully."); break
|
||
except Exception as e:
|
||
errors += len(batch); print(f"{tag} download error: {e}")
|
||
time.sleep(max(0.2, sleep_between)); continue
|
||
|
||
try:
|
||
with con: # one tx per batch
|
||
for sym in batch:
|
||
try:
|
||
total_rows += upsert_intraday(con, sym, df)
|
||
except Exception as e:
|
||
errors += 1; print(f"{tag} upsert error {sym}: {e}")
|
||
except KeyboardInterrupt:
|
||
print("\n[interrupt] stopping mid-batch."); break
|
||
|
||
print(stamp(f"{tag} done; cumulative rows={total_rows}, errors={errors}"), flush=True)
|
||
time.sleep(max(0.2, sleep_between))
|
||
|
||
print(stamp("INTRADAY END"))
|
||
print(f"[INTRADAY] rows upserted: {total_rows}, errors: {errors}")
|
||
return total_rows, errors
|
||
|
||
|
||
def _daily_count(con, sym: str) -> int:
|
||
row = con.execute("SELECT COUNT(*) FROM prices_daily WHERE symbol=?", (sym,)).fetchone()
|
||
return int(row[0] or 0)
|
||
|
||
def run_daily_incremental(con: sqlite3.Connection, symbols: List[str], period: str) -> Tuple[int,int]:
|
||
print(stamp(f"DAILY START (period={period})"))
|
||
total, errors = 0, 0
|
||
for i, sym in enumerate(symbols, 1):
|
||
try:
|
||
have = _daily_count(con, sym)
|
||
need_seed = have < 10
|
||
last = None if need_seed else get_last_daily_date(con, sym)
|
||
|
||
if last is None:
|
||
mode = "FULL"
|
||
df = yf.download(sym, period=period, interval="1d",
|
||
auto_adjust=False, progress=False)
|
||
else:
|
||
mode = f"INC from {last} -5d"
|
||
start_dt = (datetime.strptime(last, "%Y-%m-%d") - timedelta(days=5)).date().isoformat()
|
||
df = yf.download(sym, start=start_dt, interval="1d",
|
||
auto_adjust=False, progress=False)
|
||
|
||
if df is None or df.empty:
|
||
print(f"[DAILY {mode}] {sym}: yfinance returned EMPTY (have={have})")
|
||
else:
|
||
i0 = pd.to_datetime(df.index, errors="coerce")
|
||
try: i0 = i0.tz_localize(None)
|
||
except Exception: pass
|
||
dmin = str(pd.to_datetime(i0.min()).date()) if len(i0) else "NA"
|
||
dmax = str(pd.to_datetime(i0.max()).date()) if len(i0) else "NA"
|
||
print(f"[DAILY {mode}] {sym}: fetched {len(df)} rows ({dmin} → {dmax}), have={have}")
|
||
|
||
wrote = upsert_daily(con, sym, df)
|
||
total += wrote
|
||
|
||
if i % 50 == 0:
|
||
print(stamp(f"DAILY progress {i}/{len(symbols)} (rows={total})"), flush=True)
|
||
|
||
except Exception as e:
|
||
errors += 1
|
||
print(f"[daily error] {sym}: {e}")
|
||
time.sleep(0.02)
|
||
con.commit()
|
||
print(stamp("DAILY END"))
|
||
print(f"[DAILY] rows upserted: {total}, errors: {errors}")
|
||
return total, errors
|
||
|
||
|
||
# ---------- single-symbol seeder (CLI) ----------
|
||
# ---------- main loop ----------
|
||
def main():
|
||
print(stamp(f"BOOT — DB={DB_PATH}"))
|
||
con = connect_rw(DB_PATH)
|
||
try:
|
||
ensure_prices_daily_schema(con)
|
||
ensure_intraday_schema(con)
|
||
ensure_views(con, force_fix_cst=FORCE_FIX_CST)
|
||
|
||
symbols, sector_map = build_ticker_universe(SEGMENTS)
|
||
ensure_meta(con)
|
||
upsert_universe(con, symbols, sector_map)
|
||
print(f"Total tickers in universe: {len(symbols)}")
|
||
|
||
if not LOOP:
|
||
run_intraday_batched(con, symbols, LOOKBACK_DAYS_INTRADAY, BATCH_SIZE, SLEEP_BETWEEN_BATCHES)
|
||
run_daily_incremental(con, symbols, DAILY_PERIOD)
|
||
print(stamp("ONE-SHOT DONE"))
|
||
return 0
|
||
|
||
last_daily = 0.0
|
||
cycle = 0
|
||
while True:
|
||
cycle += 1
|
||
print("\n" + "="*7 + f" LOOP cycle {cycle} " + "="*7)
|
||
print(stamp("LOOP START"))
|
||
|
||
run_intraday_batched(con, symbols, LOOKBACK_DAYS_INTRADAY, BATCH_SIZE, SLEEP_BETWEEN_BATCHES)
|
||
|
||
now = time.time()
|
||
if DAILY_INTERVAL_MIN == 0 or now - last_daily >= DAILY_INTERVAL_MIN * 60:
|
||
run_daily_incremental(con, symbols, DAILY_PERIOD)
|
||
last_daily = time.time()
|
||
|
||
if INTRADAY_INTERVAL_MIN <= 0:
|
||
time.sleep(1.0)
|
||
else:
|
||
interval = INTRADAY_INTERVAL_MIN * 60
|
||
now = time.time()
|
||
next_ts = (int(now // interval) + 1) * interval
|
||
jitter = random.uniform(0, 4)
|
||
delay = max(0.0, (next_ts - now) + jitter)
|
||
print(stamp(f"SLEEP {delay:.1f}s until next intraday run"))
|
||
time.sleep(delay)
|
||
|
||
finally:
|
||
con.close()
|
||
print(stamp("CLOSED DB"))
|
||
|
||
|
||
# ---------- CLI entry ----------
|
||
def cli_entry():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="Load prices to SQLite (daily + intraday)")
|
||
parser.add_argument("--only", help="Run only for one symbol (e.g. --only ASTS)")
|
||
parser.add_argument("--daily", action="store_true", help="Run only daily for the symbol (with --only)")
|
||
parser.add_argument("--intraday", action="store_true", help="Run only intraday for the symbol (with --only)")
|
||
parser.add_argument("--diag", action="store_true", help="Diagnostic: print ranges and write CSV previews")
|
||
args = parser.parse_args()
|
||
|
||
print(stamp(f"BOOT — DB={DB_PATH}"))
|
||
con = connect_rw(DB_PATH)
|
||
ensure_prices_daily_schema(con)
|
||
ensure_intraday_schema(con)
|
||
ensure_views(con, force_fix_cst=FORCE_FIX_CST)
|
||
|
||
if args.only:
|
||
sym = args.only.strip().upper()
|
||
if args.daily:
|
||
seed_single_daily(con, sym, DAILY_PERIOD, diag=args.diag)
|
||
elif args.intraday:
|
||
# diag not needed for intraday path
|
||
run_intraday_batched(con, [sym], LOOKBACK_DAYS_INTRADAY, 1, 0.1)
|
||
else:
|
||
seed_single_daily(con, sym, DAILY_PERIOD, diag=args.diag)
|
||
run_intraday_batched(con, [sym], LOOKBACK_DAYS_INTRADAY, 1, 0.1)
|
||
con.close()
|
||
print(stamp("DONE (CLI one-shot)"))
|
||
sys.exit(0)
|
||
|
||
# Default = full loop
|
||
rc = main()
|
||
sys.exit(rc if isinstance(rc, int) else 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
cli_entry() |