1"""`findata` operator CLI."""2from __future__ import annotations34import json5from pathlib import Path6from typing import Optional78import typer910app = typer.Typer(no_args_is_help=True, help="FinData OS operator CLI")11ingest_app = typer.Typer(no_args_is_help=True)12app.add_typer(ingest_app, name="ingest")131415@app.command()16def migrate() -> None:17 """Apply database migrations."""18 from findata_core.db import migrate as _m19 applied = _m()20 typer.echo(f"applied: {applied or 'nothing'}")212223@app.command()24def discover(source: str = "all") -> None:25 """Sanitized live probe of vendor capabilities (no bulk download)."""26 out = {}27 if source in ("all", "firstrate"):28 from findata_market_firstrate import FirstRateMarketConnector29 out["firstrate"] = FirstRateMarketConnector().discover()30 if source in ("all", "firstrate_fundamentals"):31 from findata_fundamentals import FirstRateFundamentalsConnector32 out["firstrate_fundamentals"] = FirstRateFundamentalsConnector().discover()33 if source in ("all", "sec_edgar"):34 from findata_fundamentals import SecEdgarConnector35 out["sec_edgar"] = SecEdgarConnector().status()36 typer.echo(json.dumps(out, indent=2, default=str))373839@app.command()40def worker(once: bool = False, no_scheduler: bool = False, kinds: Optional[str] = None) -> None:41 """Run the single ingest worker (+ scheduler)."""42 from findata_ingest.worker import main43 main(once=once, kinds=kinds.split(",") if kinds else None, with_scheduler=not no_scheduler)444546@ingest_app.command("metadata")47def ingest_metadata(types: str = "stock,etf") -> None:48 from findata_ingest import queue49 typer.echo(queue.enqueue("metadata", None, {"types": types.split(",")}, priority=10, created_by="cli"))505152@ingest_app.command("incremental")53def ingest_incremental(type_: str = typer.Option("stock", "--type"), period: str = "day", timeframe: str = "1min",54 adjustment: str = "UNADJUSTED", force: bool = False) -> None:55 from findata_ingest import queue56 from findata_ingest.tasks.common import DATASET_ASSET57 typer.echo(queue.enqueue("incremental", DATASET_ASSET[type_], {"type": type_, "period": period, "timeframe": timeframe,58 "adjustment": adjustment, "force": force}, priority=20, created_by="cli"))596061@ingest_app.command("backfill")62def ingest_backfill(ticker_range: str, type_: str = typer.Option("stock", "--type"), timeframe: str = "1min",63 adjustment: str = "UNADJUSTED", symbols: Optional[str] = None, max_symbols: int = 0) -> None:64 """Bounded backfill of one ticker_range letter (full archive). Disk budget enforced by the lake."""65 from findata_ingest import queue66 from findata_ingest.tasks.common import DATASET_ASSET67 for letter in ticker_range.upper():68 jid = queue.enqueue("backfill", DATASET_ASSET[type_], {"type": type_, "ticker_range": letter, "timeframe": timeframe, "adjustment": adjustment,69 "symbols": symbols.split(",") if symbols else None, "max_symbols": max_symbols}, priority=35, created_by="cli",70 dedupe_key=f"backfill:{type_}:{letter}:{timeframe}:{adjustment}")71 typer.echo(f"{letter}: {jid or 'already queued'}")727374@ingest_app.command("fundamentals")75def ingest_fundamentals(tickers: str) -> None:76 from findata_ingest import queue77 typer.echo(queue.enqueue("fundamentals", "fundamentals.financials", {"tickers": tickers.split(",")}, priority=30, created_by="cli"))787980@ingest_app.command("derive")81def ingest_derive(dates: Optional[str] = None, source: str = "market_bars.stock.1m.raw") -> None:82 from findata_ingest import queue83 typer.echo(queue.enqueue("derive", source.replace(".1m.", ".5m."), {"source_dataset": source, "target_interval": "5m", "dates": dates.split(",") if dates else []}, created_by="cli"))848586@ingest_app.command("compaction")87def ingest_compaction(dataset: str = "market_bars.stock.1m.raw", min_age_days: int = 35) -> None:88 from findata_ingest import queue89 typer.echo(queue.enqueue("compaction", dataset, {"min_age_days": min_age_days}, priority=60, created_by="cli"))909192@ingest_app.command("jobs")93def ingest_jobs(limit: int = 20) -> None:94 from findata_core.db import fetch_all95 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,)):96 typer.echo(f"{str(r['id'])[:8]} {r['kind']:<12} {str(r['dataset_id']):<28} {r['status']:<9} a={r['attempts']} {r['created_at']:%m-%d %H:%M} {r['err'] or ''}")979899@app.command()100def backup(now: bool = True) -> None:101 from findata_ingest import queue102 typer.echo(queue.enqueue("backup", None, {}, priority=5, created_by="cli"))103104105@app.command("restore-drill")106def restore_drill(dest: Path = Path("/tmp/findata-restore-drill"), dataset: str = "market_bars.stock.1m.raw") -> None:107 from findata_core.config import get_settings108 from findata_ingest.tasks.backup import restore_drill as rd109 typer.echo(json.dumps(rd(get_settings().backup_target, dest, dataset), indent=2, default=str))110111112@app.command("create-owner")113def create_owner(email: str, password: Optional[str] = None) -> None:114 """Create (or reset the password of) the owner account; prints the TOTP provisioning URI."""115 from findata_api.auth import create_owner as _co116 typer.echo(json.dumps(_co(email, password), indent=2))117118119@app.command("create-key")120def create_key(email: str, name: str, scopes: str = "read:market,read:fundamentals,query,export") -> None:121 from findata_api.auth import create_api_key_for_email122 typer.echo(json.dumps(create_api_key_for_email(email, name, scopes.split(",")), indent=2))123124125@app.command()126def status() -> None:127 from findata_core.db import fetch_all, fetch_one128 from findata_ingest.lake import Lake129 lake = Lake()130 typer.echo(f"lake {lake.root} free {lake.free_gb():.1f} GB curated {lake.used_bytes('curated')/1e9:.2f} GB raw {lake.used_bytes('raw')/1e9:.2f} GB")131 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"):132 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")133 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")134 typer.echo(f"jobs queued={q['q']} running={q['r']} dead={q['d']}")135136137if __name__ == "__main__":138 app()139