1"""`findata` operator CLI."""2from __future__ import annotations34import json5from pathlib import Path67import typer89app = typer.Typer(no_args_is_help=True, help="FinData OS operator CLI")101112@app.callback()13def _main() -> None:14 import atexit1516 from findata_core.db import close_pool17 atexit.register(close_pool)18ingest_app = typer.Typer(no_args_is_help=True)19app.add_typer(ingest_app, name="ingest")202122@app.command()23def migrate() -> None:24 """Apply database migrations."""25 from findata_core.db import migrate as _m26 applied = _m()27 typer.echo(f"applied: {applied or 'nothing'}")282930@app.command()31def discover(source: str = "all") -> None:32 """Sanitized live probe of vendor capabilities (no bulk download)."""33 out = {}34 if source in ("all", "firstrate"):35 from findata_market_firstrate import FirstRateMarketConnector36 out["firstrate"] = FirstRateMarketConnector().discover()37 if source in ("all", "firstrate_fundamentals"):38 from findata_fundamentals import FirstRateFundamentalsConnector39 out["firstrate_fundamentals"] = FirstRateFundamentalsConnector().discover()40 if source in ("all", "sec_edgar"):41 from findata_fundamentals import SecEdgarConnector42 out["sec_edgar"] = SecEdgarConnector().status()43 typer.echo(json.dumps(out, indent=2, default=str))444546@app.command()47def worker(once: bool = False, no_scheduler: bool = False, kinds: str | None = None, role: str = "primary") -> None:48 """Run an ingest worker: role=primary (lake lock + scheduler) or role=bulk (parallel backfill/options/fundamentals)."""49 from findata_ingest.worker import main50 main(once=once, kinds=kinds.split(",") if kinds else None, with_scheduler=not no_scheduler, role=role)515253@ingest_app.command("plan-all")54def ingest_plan_all(types: str | None = None, no_options: bool = False, no_delisted: bool = False) -> None:55 """Show the complete vendor download plan (counts per dataset), without enqueuing anything."""56 from collections import Counter5758 from findata_market_firstrate import build_full_plan59 plan = build_full_plan(types.split(",") if types else None, include_options=not no_options, include_delisted=not no_delisted)60 typer.echo(f"{len(plan)} jobs")61 for ds, n in sorted(Counter(j["dataset_id"] for j in plan).items()):62 typer.echo(f" {ds:<48} {n}")636465@ingest_app.command("backfill-all")66def ingest_backfill_all(types: str | None = None, no_options: bool = False, no_delisted: bool = False, yes: bool = False) -> None:67 """Enqueue the complete vendor download (every licensed archive). Needs the cold lake root (external SSD)."""68 from findata_core.config import get_settings69 from findata_market_firstrate import build_full_plan7071 from findata_ingest import queue72 from findata_ingest.lake import Lake73 lake = Lake()74 s = get_settings()75 free_hot = lake.free_gb()76 free_cold = lake.free_gb(lake.cold) if lake.cold else None77 typer.echo(f"hot root {lake.root}: {free_hot:.0f} GB free · cold root {lake.cold}: {free_cold if free_cold is None else round(free_cold)} GB free")78 if not lake.cold:79 typer.echo("refusing: FINDATA_LAKE_COLD is not configured (the full catalog needs the external SSD)")80 raise typer.Exit(2)81 if free_cold < 1200 or free_hot < 300:82 typer.echo("refusing: need >= 1200 GB free on the cold root and >= 300 GB on the hot root for the full catalog")83 raise typer.Exit(2)84 plan = build_full_plan(types.split(",") if types else None, include_options=not no_options, include_delisted=not no_delisted)85 if not yes and not typer.confirm(f"enqueue {len(plan)} jobs?"):86 raise typer.Exit(1)87 n = 088 for j in plan:89 if queue.enqueue(j["kind"], j["dataset_id"], j["params"], j["priority"], dedupe_key=j["dedupe_key"], created_by="cli:backfill-all", max_attempts=4):90 n += 191 typer.echo(f"enqueued {n} new jobs ({len(plan) - n} already queued/running); bulk workers configured: {s.bulk_workers}")929394@ingest_app.command("options")95def ingest_options(year: int | None = None, quarter: str | None = None, month: bool = False) -> None:96 from findata_ingest import queue97 params = {"period": "month"} if month else {"year": year, "quarter": quarter}98 typer.echo(queue.enqueue("options", "options.eod", params, priority=55, created_by="cli"))99100101@ingest_app.command("metadata")102def ingest_metadata(types: str = "stock,etf") -> None:103 from findata_ingest import queue104 typer.echo(queue.enqueue("metadata", None, {"types": types.split(",")}, priority=10, created_by="cli"))105106107@ingest_app.command("incremental")108def ingest_incremental(type_: str = typer.Option("stock", "--type"), period: str = "day", timeframe: str = "1min",109 adjustment: str = "UNADJUSTED", force: bool = False) -> None:110 from findata_ingest import queue111 from findata_ingest.tasks.common import DATASET_ASSET112 typer.echo(queue.enqueue("incremental", DATASET_ASSET[type_], {"type": type_, "period": period, "timeframe": timeframe,113 "adjustment": adjustment, "force": force}, priority=20, created_by="cli"))114115116@ingest_app.command("backfill")117def ingest_backfill(ticker_range: str, type_: str = typer.Option("stock", "--type"), timeframe: str = "1min",118 adjustment: str = "UNADJUSTED", symbols: str | None = None, max_symbols: int = 0) -> None:119 """Bounded backfill of one ticker_range letter (full archive). Disk budget enforced by the lake."""120 from findata_ingest import queue121 from findata_ingest.tasks.common import DATASET_ASSET122 for letter in ticker_range.upper():123 jid = queue.enqueue("backfill", DATASET_ASSET[type_], {"type": type_, "ticker_range": letter, "timeframe": timeframe, "adjustment": adjustment,124 "symbols": symbols.split(",") if symbols else None, "max_symbols": max_symbols}, priority=35, created_by="cli",125 dedupe_key=f"backfill:{type_}:{letter}:{timeframe}:{adjustment}")126 typer.echo(f"{letter}: {jid or 'already queued'}")127128129@ingest_app.command("fundamentals")130def ingest_fundamentals(tickers: str | None = typer.Argument(None), all_active: bool = False) -> None:131 """Fundamentals for a ticker list, or --all-active for every active US stock (checkpointed per ticker)."""132 from findata_ingest import queue133 params = {"all_active": True} if all_active else {"tickers": (tickers or "").split(",")}134 typer.echo(queue.enqueue("fundamentals", "fundamentals.financials", params, priority=58, created_by="cli",135 dedupe_key="fundamentals:all_active" if all_active else None))136137138@ingest_app.command("derive")139def ingest_derive(dates: str | None = None, source: str = "market_bars.stock.1m.raw") -> None:140 from findata_ingest import queue141 typer.echo(queue.enqueue("derive", source.replace(".1m.", ".5m."), {"source_dataset": source, "target_interval": "5m", "dates": dates.split(",") if dates else []}, created_by="cli"))142143144@ingest_app.command("compaction")145def ingest_compaction(dataset: str = "market_bars.stock.1m.raw", min_age_days: int = 35) -> None:146 from findata_ingest import queue147 typer.echo(queue.enqueue("compaction", dataset, {"min_age_days": min_age_days}, priority=60, created_by="cli"))148149150@ingest_app.command("jobs")151def ingest_jobs(limit: int = 20) -> None:152 from findata_core.db import fetch_all153 for r in fetch_all("SELECT id, kind, dataset_id, status, attempts, created_at, finished_at, left(last_error, 80) err FROM ingest_jobs ORDER BY created_at DESC LIMIT %s", (limit,)):154 typer.echo(f"{str(r['id'])[:8]} {r['kind']:<12} {r['dataset_id']!s:<28} {r['status']:<9} a={r['attempts']} {r['created_at']:%m-%d %H:%M} {r['err'] or ''}")155156157@app.command()158def backup(now: bool = True) -> None:159 from findata_ingest import queue160 typer.echo(queue.enqueue("backup", None, {}, priority=5, created_by="cli"))161162163@app.command("restore-drill")164def restore_drill(dest: Path = Path("/tmp/findata-restore-drill"), dataset: str = "market_bars.stock.1m.raw") -> None:165 from findata_core.config import get_settings166167 from findata_ingest.tasks.backup import restore_drill as rd168 typer.echo(json.dumps(rd(get_settings().backup_target, dest, dataset), indent=2, default=str))169170171@app.command("create-owner")172def create_owner(email: str, password: str | None = None) -> None:173 """Create (or reset the password of) the owner account; prints the TOTP provisioning URI."""174 from findata_api.auth import create_owner as _co175 typer.echo(json.dumps(_co(email, password), indent=2))176177178@app.command("create-key")179def create_key(email: str, name: str, scopes: str = "read:market,read:fundamentals,query,export") -> None:180 from findata_api.auth import create_api_key_for_email181 typer.echo(json.dumps(create_api_key_for_email(email, name, scopes.split(",")), indent=2))182183184@app.command()185def manifest(dataset: str, version: int | None = None, out: Path | None = None) -> None:186 """Write the full file manifest of a dataset version (reconstructed from lake_files) as JSON."""187 from findata_ingest.lake import Lake188 files = Lake().snapshot_files(dataset, version)189 doc = {"dataset_id": dataset, "version": version, "file_count": len(files), "rows": sum(f["rows"] for f in files), "files": files}190 text = json.dumps(doc, indent=1)191 if out:192 out.write_text(text)193 typer.echo(f"{out} ({len(files)} files)")194 else:195 typer.echo(text)196197198@app.command()199def status() -> None:200 from findata_core.db import fetch_all, fetch_one201202 from findata_ingest.lake import Lake203 lake = Lake()204 u = fetch_one("SELECT coalesce(sum(bytes) FILTER (WHERE version_removed IS NULL),0) AS live FROM lake_files")205 r = fetch_one("SELECT coalesce(sum(bytes),0) AS raw FROM raw_archives WHERE path IS NOT NULL")206 cold = f" cold {lake.cold} free {lake.free_gb(lake.cold):.0f} GB" if lake.cold else ""207 typer.echo(f"lake {lake.root} free {lake.free_gb():.1f} GB curated {u['live']/1e9:.2f} GB raw {r['raw']/1e9:.2f} GB{cold}")208 for r in fetch_all("SELECT d.id, d.status, d.current_version, v.file_count, v.row_count, v.bytes FROM datasets d LEFT JOIN dataset_versions v ON v.dataset_id=d.id AND v.version=d.current_version ORDER BY d.id"):209 typer.echo(f" {r['id']:<36} {r['status']:<22} v{r['current_version']:<4} files={r['file_count'] or 0:<6} rows={r['row_count'] or 0:<12} {(r['bytes'] or 0)/1e6:.1f} MB")210 q = fetch_one("SELECT count(*) FILTER (WHERE status='queued') q, count(*) FILTER (WHERE status='running') r, count(*) FILTER (WHERE status='dead') d FROM ingest_jobs")211 typer.echo(f"jobs queued={q['q']} running={q['r']} dead={q['d']}")212213214if __name__ == "__main__":215 app()216