from __future__ import annotations import argparse from decimal import Decimal from pathlib import Path from jarvis_finance.config.settings import ensure_runtime_dirs, load_settings from jarvis_finance.fx.providers import FrankfurterFxProvider, MockFxProvider, TwelveDataFxProvider from jarvis_finance.fx.rates import recheck_transaction_fx_status, update_fx_rates from jarvis_finance.imports.instrument_candidates import qwen_resolve_staged_candidates, stage_docx_candidates from jarvis_finance.market.providers import CoinGeckoClient, refresh_crypto_prices from jarvis_finance.market_data.candidates import CompositePublicLookupProvider, EmptyLookupProvider, dedupe_missing_provider_symbol_alerts, export_mapping_decision_pack, export_mapping_review_template, generate_mapping_candidates, recalculate_candidate_rankings from jarvis_finance.market_data.catalog import prepare_catalog_from_instruments from jarvis_finance.market_data.prices import MockEquityPriceProvider, equity_price_provider_by_name, refresh_market_prices from jarvis_finance.market_data.cache import backfill_crypto_points_from_crypto_prices, backfill_equity_points_from_market_prices from jarvis_finance.quality.data_quality import check_crypto_data_quality from jarvis_finance.quality.git_safety import assert_safe, scan_path from jarvis_finance.runtime.backup import backup_runtime_db, restore_runtime_db, verify_backup from jarvis_finance.storage.database import connect from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="finance") sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("init-db") sub.add_parser("migrate") scan = sub.add_parser("git-safety-scan") scan.add_argument("path", nargs="?", default=".") prices = sub.add_parser("update-crypto-prices") prices.add_argument("--currency", default="CHF") prices.add_argument("--max-age-seconds", type=int, default=3600) prices.add_argument("--sleep-seconds", type=float, default=0.0) prices.add_argument("--max-retries", type=int, default=2) prices.add_argument("--initial-backoff", type=float, default=1.0) prices.add_argument("--max-backoff", type=float, default=8.0) prices.add_argument("--only-missing", action="store_true") prices.add_argument("--only-stale", action="store_true") prices.add_argument("--only-symbol") prices.add_argument("--limit", type=int) prices.add_argument("--dry-run", action="store_true") dq = sub.add_parser("check-data-quality") dq.add_argument("--scope", choices=["crypto"], default="crypto") dq.add_argument("--resolve-fixed", action="store_true") dq.add_argument("--dry-run", action="store_true") dq.add_argument("--currency", default="CHF") dq.add_argument("--max-age-seconds", type=int, default=86_400) fx = sub.add_parser("update-fx-rates") fx.add_argument("--date", dest="rate_date") fx.add_argument("--latest", action="store_true") fx.add_argument("--missing-only", action="store_true") fx.add_argument("--dry-run", action="store_true") fx.add_argument("--currency", action="append", dest="currencies") fx.add_argument("--provider", choices=["frankfurter", "twelvedata", "mock"], default="frankfurter") fx.add_argument("--mock-rate", action="append", default=[], help="Test-only BASE=RATE, e.g. USD=0.9") fx.add_argument("--recheck-transactions", action="store_true") fx.add_argument("--source-type") market_prices = sub.add_parser("update-market-prices") market_prices.add_argument("--asset-class", choices=["etf", "equity", "stock"], required=True) market_prices.add_argument("--only-missing", action="store_true") market_prices.add_argument("--only-stale", action="store_true") market_prices.add_argument("--only-isin") market_prices.add_argument("--limit", type=int) market_prices.add_argument("--dry-run", action="store_true") market_prices.add_argument("--date", dest="price_date") market_prices.add_argument("--mock-price", action="append", default=[], help="Test-only SYMBOL=PRICE") equity_prices = sub.add_parser("update-equity-prices") equity_prices.add_argument("--asset-class", choices=["etf", "equity", "stock", "all"], default="all") equity_prices.add_argument("--only-missing", action="store_true") equity_prices.add_argument("--only-stale", action="store_true") equity_prices.add_argument("--only-isin") equity_prices.add_argument("--limit", type=int) equity_prices.add_argument("--dry-run", action="store_true") equity_prices.add_argument("--date", dest="price_date") equity_prices.add_argument("--provider", choices=["auto", "fmp", "twelvedata", "finnhub", "massive", "eodhd", "mock"], default="auto") equity_prices.add_argument("--allow-low-volume", action="store_true", help="Allow explicit low-volume providers such as EODHD") equity_prices.add_argument("--mock-price", action="append", default=[], help="Test-only SYMBOL=PRICE") equity_quotes = sub.add_parser("update-equity-quotes") equity_quotes.add_argument("--provider", choices=["auto", "fmp", "twelvedata", "finnhub", "massive", "eodhd", "mock"], default="auto") equity_quotes.add_argument("--limit", type=int, default=10) equity_quotes.add_argument("--dry-run", action="store_true") equity_quotes.add_argument("--mock-price", action="append", default=[], help="Test-only SYMBOL=PRICE") crypto_live = sub.add_parser("update-crypto-live-stats") crypto_live.add_argument("--provider", choices=["coingecko", "binance"], default="binance") crypto_live.add_argument("--limit", type=int, default=20) crypto_live.add_argument("--dry-run", action="store_true") charts = sub.add_parser("update-market-charts") charts.add_argument("--asset-class", choices=["equity", "crypto"], required=True) charts.add_argument("--range", default="1d") charts.add_argument("--interval", default="5m") charts.add_argument("--dry-run", action="store_true") catalog = sub.add_parser("prepare-instrument-catalog") catalog.add_argument("--source", default="true_wealth") catalog.add_argument("--limit", type=int) candidates = sub.add_parser("generate-mapping-candidates") candidates.add_argument("--source", default="true_wealth") candidates.add_argument("--limit", type=int) candidates.add_argument("--dry-run", action="store_true") candidates.add_argument("--public-lookup", action="store_true") candidates.add_argument("--export-template", action="store_true") candidates.add_argument("--template-path") candidates.add_argument("--dedupe-missing-provider-alerts", action="store_true") rank_candidates = sub.add_parser("rank-mapping-candidates") rank_candidates.add_argument("--export-decision-pack", action="store_true") rank_candidates.add_argument("--decision-pack-path") rank_candidates.add_argument("--top-n", type=int, default=5) q = sub.add_parser("stage-instrument-candidates") q.add_argument("files", nargs="+", help="Broker DOCX files outside git repo") q.add_argument("--qwen", action="store_true") q.add_argument("--dry-run", action="store_true") q.add_argument("--allow-repo-files", action="store_true") qr = sub.add_parser("qwen-resolve-instrument-candidates") qr.add_argument("--endpoint", default="http://100.101.173.25:11435/v1") qr.add_argument("--model", default="qwen-3.6-agent") qr.add_argument("--limit", type=int) qr.add_argument("--dry-run", action="store_true") qr.add_argument("--timeout", type=int, default=45) sub.add_parser("backup-runtime-db") restore = sub.add_parser("restore-runtime-db") restore.add_argument("--from", dest="backup_file", required=True) restore.add_argument("--yes", action="store_true") verify = sub.add_parser("verify-backup") verify.add_argument("--file", required=True) args = parser.parse_args(argv) if args.command in {"init-db", "migrate"}: settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) apply_migrations(conn) print(f"SCHEMA_VERSION={get_schema_version(conn)}") return 0 if args.command == "git-safety-scan": findings = scan_path(Path(args.path)) if findings: for finding in findings: print(f"{finding.path}: {finding.reason}") return 1 assert_safe(Path(args.path)) print("GIT_SAFETY_OK") return 0 if args.command == "update-crypto-prices": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) result = refresh_crypto_prices( conn, provider=CoinGeckoClient( max_retries=args.max_retries, initial_backoff_seconds=args.initial_backoff, max_backoff_seconds=args.max_backoff, ), currency=args.currency, max_age_seconds=args.max_age_seconds, only_missing=args.only_missing, only_stale=args.only_stale, only_symbol=args.only_symbol, limit=args.limit, dry_run=args.dry_run, sleep_seconds=args.sleep_seconds, ) print( "CRYPTO_PRICE_UPDATE " f"currency={result.currency} total_assets={result.total_assets} " f"updated={result.updated_count} cached={result.cached_count} " f"skipped={result.skipped_count} stale={result.stale_count} " f"warnings={result.warning_count} errors={result.error_count} " f"assets_still_missing_local_price={result.missing_local_price_count} " f"dry_run={str(result.dry_run).lower()}" ) return 1 if result.error_count else 0 finally: conn.close() if args.command == "update-fx-rates": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) if args.recheck_transactions: result = recheck_transaction_fx_status(conn, source_type=args.source_type, dry_run=args.dry_run) print( "FX_RECHECK " f"transactions={result.transaction_count} chf_positions={result.chf_positions} " f"foreign_currency_positions={result.foreign_currency_positions} " f"justified_missing_fx_alerts={result.justified_missing_fx_alerts} " f"false_positive_missing_fx_alerts={result.false_positive_missing_fx_alerts} " f"corrected_transactions={result.corrected_transactions} resolved_alerts={result.resolved_alerts} " f"audit_events={result.audit_events} dry_run={str(result.dry_run).lower()}" ) return 0 if not args.rate_date and not args.latest: parser.error("update-fx-rates requires --date or --latest unless --recheck-transactions is used") mock_rates: dict[tuple[str, str, str], Decimal | None] = {} for item in args.mock_rate: base, rate = item.split("=", 1) mock_rates[(base.upper(), "CHF", args.rate_date or "latest")] = Decimal(rate) currencies = args.currencies or sorted({r["currency_original"] for r in conn.execute("SELECT DISTINCT currency_original FROM transactions WHERE currency_original IS NOT NULL").fetchall()}) if args.mock_rate or args.provider == "mock": provider = MockFxProvider(mock_rates) elif args.provider == "twelvedata": provider = TwelveDataFxProvider() else: provider = FrankfurterFxProvider() result = update_fx_rates(conn, provider=provider, currencies=currencies, rate_date=args.rate_date or "latest", latest=args.latest, missing_only=args.missing_only, dry_run=args.dry_run) print( "FX_RATE_UPDATE " f"updated={result.updated_count} cached={result.cached_count} skipped={result.skipped_count} " f"warnings={result.warning_count} errors={result.error_count} resolved_alerts={result.resolved_alerts} " f"dry_run={str(result.dry_run).lower()}" ) return 1 if result.error_count else 0 finally: conn.close() if args.command == "update-equity-quotes": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) prices = {} for item in args.mock_price: symbol, price = item.split("=", 1) prices[symbol.upper()] = Decimal(price) provider = MockEquityPriceProvider(prices) if prices or args.provider == "mock" else equity_price_provider_by_name(args.provider) exit_code = 0 for asset_class in ("stock", "etf"): result = refresh_market_prices(conn, provider=provider, asset_class=asset_class, limit=args.limit, dry_run=args.dry_run) print( "EQUITY_QUOTE_UPDATE " f"asset_class={result.asset_class} total_mappings={result.total_mappings} updated={result.updated_count} " f"cached={result.cached_count} skipped={result.skipped_count} stale={result.stale_count} " f"warnings={result.warning_count} errors={result.error_count} dry_run={str(result.dry_run).lower()}" ) exit_code = max(exit_code, 1 if result.error_count else 0) return exit_code finally: conn.close() if args.command == "update-crypto-live-stats": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) if args.provider == "binance": from jarvis_finance.api.schemas.market import QuoteRefreshRequest as _QReq from jarvis_finance.services.market_service import refresh_crypto_quote assets = conn.execute("SELECT asset_id FROM crypto_assets WHERE is_active=1 AND binance_symbol IS NOT NULL AND binance_symbol!='' ORDER BY symbol LIMIT ?", (args.limit,)).fetchall() updated = warnings = errors = 0 for asset in assets: response = refresh_crypto_quote(conn, asset["asset_id"], _QReq(provider="binance", dry_run=args.dry_run)) if response.quality_status == "fresh": updated += 1 elif response.quality_status in {"rate_limited", "provider_error"}: errors += 1 else: warnings += 1 print(f"CRYPTO_LIVE_STATS_UPDATE provider=binance total_assets={len(assets)} updated={updated} warnings={warnings} errors={errors} dry_run={str(args.dry_run).lower()}") return 1 if errors else 0 result = refresh_crypto_prices(conn, provider=CoinGeckoClient(), currency="USD", limit=args.limit, dry_run=args.dry_run) print(f"CRYPTO_LIVE_STATS_UPDATE provider=coingecko total_assets={result.total_assets} updated={result.updated_count} warnings={result.warning_count} errors={result.error_count} dry_run={str(result.dry_run).lower()}") return 1 if result.error_count else 0 finally: conn.close() if args.command == "update-market-charts": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) written = 0 if not args.dry_run and args.asset_class == "equity": for row in conn.execute("SELECT DISTINCT instrument_id FROM market_prices").fetchall(): written += backfill_equity_points_from_market_prices(conn, row["instrument_id"]) elif not args.dry_run: for row in conn.execute("SELECT DISTINCT asset_id FROM crypto_prices").fetchall(): written += backfill_crypto_points_from_crypto_prices(conn, row["asset_id"], "CHF") print(f"MARKET_CHART_UPDATE asset_class={args.asset_class} points_cached={written} range={args.range} interval={args.interval} dry_run={str(args.dry_run).lower()}") return 0 finally: conn.close() if args.command in {"update-market-prices", "update-equity-prices"}: settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) prices = {} for item in args.mock_price: symbol, price = item.split("=", 1) prices[symbol.upper()] = Decimal(price) if args.command == "update-market-prices": provider = MockEquityPriceProvider(prices) elif prices or args.provider == "mock": provider = MockEquityPriceProvider(prices) elif args.provider == "eodhd" and not args.allow_low_volume: parser.error("EODHD is low-volume; pass --provider eodhd --allow-low-volume explicitly") else: provider = equity_price_provider_by_name(args.provider) asset_classes = ["stock", "etf"] if args.command == "update-equity-prices" and args.asset_class == "all" else ["stock" if args.asset_class == "equity" else args.asset_class] totals = [] exit_code = 0 for asset_class in asset_classes: result = refresh_market_prices( conn, provider=provider, asset_class=asset_class, price_date=args.price_date, only_missing=args.only_missing, only_stale=args.only_stale, only_isin=args.only_isin, limit=args.limit, dry_run=args.dry_run, ) totals.append(result) exit_code = max(exit_code, 1 if result.error_count else 0) print( ("EQUITY_PRICE_UPDATE " if args.command == "update-equity-prices" else "MARKET_PRICE_UPDATE ") + f"asset_class={result.asset_class} total_mappings={result.total_mappings} updated={result.updated_count} " f"cached={result.cached_count} skipped={result.skipped_count} stale={result.stale_count} " f"warnings={result.warning_count} errors={result.error_count} dry_run={str(result.dry_run).lower()}" ) return exit_code finally: conn.close() if args.command == "prepare-instrument-catalog": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) rows = conn.execute( """ SELECT DISTINCT t.instrument_id FROM transactions t JOIN instruments i ON i.instrument_id=t.instrument_id WHERE t.transaction_type='initial_position_snapshot' AND t.source_type='broker_import_reviewed_snapshot' AND COALESCE(t.is_voided,0)=0 AND t.instrument_id IS NOT NULL ORDER BY i.isin, i.name """ ).fetchall() ids = [r["instrument_id"] for r in rows] if args.limit is not None: ids = ids[:args.limit] summary = prepare_catalog_from_instruments(conn, instrument_ids=ids, source=args.source) print( "INSTRUMENT_CATALOG_PREP " f"instruments_total={summary['instruments_total']} catalog_entries_created={summary['catalog_entries_created']} " f"multiple_listings={summary['multiple_listings']} manual_review_required={summary['manual_review_required']} " f"hedge_unknown={summary['hedge_unknown']} instrument_status_unknown={summary['instrument_status_unknown']} " f"valuation_ready={summary['valuation_ready']}" ) return 0 finally: conn.close() if args.command == "generate-mapping-candidates": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) if args.dedupe_missing_provider_alerts: resolved = dedupe_missing_provider_symbol_alerts(conn) else: resolved = 0 provider = CompositePublicLookupProvider() if args.public_lookup else EmptyLookupProvider() result = generate_mapping_candidates(conn, provider=provider, source=args.source, limit=args.limit, dry_run=args.dry_run) template = None if args.export_template: template_path = args.template_path or str(settings.runtime_paths.exports_dir / "true_wealth_mapping_review_template.csv") template = export_mapping_review_template(conn, output_path=template_path) print( "MAPPING_CANDIDATES " f"instruments_total={result.instruments_total} candidates_found={result.candidates_found} " f"high={result.high_confidence} medium={result.medium_confidence} low={result.low_confidence} " f"instruments_without_candidate={result.instruments_without_candidate} " f"hedge_unknown={result.instruments_with_hedge_unknown} instrument_status_unknown={result.instruments_with_status_unknown} " f"provider_lookup_unavailable={result.provider_lookup_unavailable} ambiguous={result.ambiguous_provider_mapping} " f"deduped_missing_provider_alerts={resolved} " f"template_updated={str(bool(template)).lower()} template_instruments={template['instrument_count'] if template else 0} template_candidates={template['candidate_rows'] if template else 0} " f"dry_run={str(result.dry_run).lower()}" ) return 0 finally: conn.close() if args.command == "rank-mapping-candidates": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) summary = recalculate_candidate_rankings(conn) decision = None if args.export_decision_pack: decision_path = args.decision_pack_path or str(settings.runtime_paths.exports_dir / "mapping_candidate_decision_pack.md") decision = export_mapping_decision_pack(conn, output_path=decision_path, top_n=args.top_n) print( "MAPPING_CANDIDATE_RANKING " f"candidates_total={summary['candidates_total']} rank_high={summary['rank_high']} " f"rank_medium={summary['rank_medium']} rank_low={summary['rank_low']} " f"multiple_listings_same_isin={summary['multiple_listings_same_isin']} " f"currency_mismatch={summary['currency_mismatch']} hedge_status_unknown={summary['hedge_status_unknown']} " f"instrument_status_unknown={summary['instrument_status_unknown']} " f"candidate_review_required={summary['candidate_review_required']} " f"decision_pack_created={str(bool(decision)).lower()} " f"decision_pack_instruments={decision['instrument_count'] if decision else 0} " f"decision_pack_candidates={decision['candidate_rows'] if decision else 0} " f"top_candidates_present={str(bool(decision and decision['top_candidates_present'])).lower()}" ) return 0 finally: conn.close() if args.command == "stage-instrument-candidates": settings = load_settings() ensure_runtime_dirs(settings) repo_root = Path.cwd().resolve() files = [Path(f).expanduser().resolve() for f in args.files] if not args.allow_repo_files: for f in files: try: f.relative_to(repo_root) parser.error("Broker files must stay outside git repo; pass runtime import paths, not repo files") except ValueError: pass conn = connect(settings.db_path) try: apply_migrations(conn) result = stage_docx_candidates(conn, files, use_qwen=args.qwen, dry_run=args.dry_run) print( "INSTRUMENT_CANDIDATE_STAGING " f"candidates_total={result.candidates_total} with_isin={result.with_isin} without_isin={result.without_isin} " f"with_ticker={result.with_ticker} with_currency={result.with_currency} " f"exact_isin_match={result.exact_isin_match} probable={result.probable} needs_manual_review={result.needs_manual_review} " f"confirmed={result.confirmed} importable_confirmed={result.importable_confirmed} blocked={result.blocked} " f"qwen_requested={str(args.qwen).lower()} qwen_used={str(result.qwen_used).lower()} qwen_available={str(result.qwen_available).lower()} " f"parser_qwen_agree={result.parser_qwen_agree} parser_qwen_disagree={result.parser_qwen_disagree} " f"warnings={result.warnings} errors={result.errors} dry_run={str(result.dry_run).lower()}" ) return 1 if result.errors else 0 finally: conn.close() if args.command == "qwen-resolve-instrument-candidates": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) result = qwen_resolve_staged_candidates(conn, endpoint=args.endpoint, model=args.model, limit=args.limit, dry_run=args.dry_run, timeout=args.timeout) print( "QWEN_INSTRUMENT_RESOLVE " f"qwen_available={str(result.qwen_available).lower()} qwen_used={str(result.qwen_used).lower()} model={result.model} " f"candidates_total={result.candidates_total} " f"exact_before={result.exact_before} exact_after={result.exact_after} " f"probable_before={result.probable_before} probable_after={result.probable_after} " f"needs_review_before={result.needs_review_before} needs_review_after={result.needs_review_after} " f"confirmed_after={result.confirmed_after} improved={result.improved} " f"manual_review_remaining={result.manual_review_remaining} warnings={result.warnings} errors={result.errors} " f"dry_run={str(result.dry_run).lower()}" ) return 1 if result.errors else 0 finally: conn.close() if args.command == "check-data-quality": settings = load_settings() ensure_runtime_dirs(settings) conn = connect(settings.db_path) try: apply_migrations(conn) result = check_crypto_data_quality( conn, currency=args.currency, max_age_seconds=args.max_age_seconds, resolve_fixed=args.resolve_fixed, dry_run=args.dry_run, ) print( "DATA_QUALITY_CHECK " f"scope={result.scope} active_crypto_assets={result.active_crypto_assets} " f"fresh_price_assets={result.fresh_price_assets} " f"active_stale_price_alerts_before={result.active_stale_price_alerts_before} " f"active_stale_price_alerts_after={result.active_stale_price_alerts_after} " f"resolved_alerts={result.resolved_alerts} " f"warnings={result.warnings} errors={result.errors} " f"dry_run={str(result.dry_run).lower()}" ) return 1 if result.errors else 0 finally: conn.close() if args.command == "backup-runtime-db": settings = load_settings() ensure_runtime_dirs(settings) result = backup_runtime_db( db_path=settings.db_path, backups_dir=settings.runtime_paths.backups_dir, repo_root=Path.cwd(), ) print( "RUNTIME_DB_BACKUP " f"backup_file={result.backup_file} checksum_file={result.checksum_file} " f"sha256={result.sha256} size_bytes={result.size_bytes}" ) return 0 if args.command == "verify-backup": result = verify_backup(backup_file=Path(args.file), repo_root=Path.cwd()) print( "BACKUP_VERIFY " f"backup_file={result.backup_file} checksum_file={result.checksum_file} " f"exists={str(result.exists).lower()} checksum_ok={str(result.checksum_ok).lower()} " f"sha256={result.sha256}" ) return 0 if result.exists and result.checksum_ok else 1 if args.command == "restore-runtime-db": settings = load_settings() ensure_runtime_dirs(settings) result = restore_runtime_db( backup_file=Path(args.backup_file), db_path=settings.db_path, repo_root=Path.cwd(), yes=args.yes, ) print( "RUNTIME_DB_RESTORE " f"source_file={result.source_file} restored_to={result.restored_to} " f"verified={str(result.verified).lower()}" ) return 0 return 2 if __name__ == "__main__": raise SystemExit(main())