from __future__ import annotations import argparse, json, re, sys, urllib.request from dataclasses import dataclass, asdict from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any from openpyxl import load_workbook REPO = Path('/home/agent/.hermes/repos/FinanceManager') sys.path.insert(0, str(REPO/'src')) from jarvis_finance.config.settings import load_settings, find_repo_root from jarvis_finance.storage.database import connect from jarvis_finance.storage.migrations import apply_migrations from jarvis_finance.imports.common import create_import_session, source_hash, utc_now, stable_id from jarvis_finance.crypto.assets import create_crypto_asset from jarvis_finance.crypto.wallets import create_wallet from jarvis_finance.crypto.holdings import create_initial_holding_snapshot from jarvis_finance.audit.log import record_audit_event XLSX = Path('/home/agent/jarvis_runtime/finance-system/tmp_imports/260514_crypto_adjusted.xlsx') VALIDATION_CACHE = Path('/home/agent/jarvis_runtime/finance-system/tmp_imports/coingecko_coins_list.json') SNAPSHOT_DATE='2025-12-31' WALLET_COLS = range(3,10) # C:I @dataclass class ParsedHolding: row_idx:int coingecko_id:str coin_label:str symbol_guess:str wallet_label:str quantity_text:str formula:bool=False scientific:bool=False @dataclass class Plan: coin_rows_read:int=0 coingecko_id_rows_detected:int=0 valid_ids:int=0 invalid_or_missing_ids:int=0 semantic_conflicts:int=0 existing_assets:int=0 new_assets_importable:int=0 existing_holdings:int=0 new_holdings_importable:int=0 mapping_conflicts:int=0 skipped:int=0 decimal_problems:int=0 formula_cases:int=0 scientific_notation_cases:int=0 wallet_missing_importable:int=0 blockers:int=0 import_possible:bool=False parsed_holdings:int=0 errors:int=0 import_session_created:bool=False audit_events_created:int=0 dashboard_crypto_contains_new_coins:bool=False report_context_contains_new_coins:bool=False assets_created:int=0 holdings_created:int=0 existing_skipped:int=0 conflicts:int=0 def load_cg() -> dict[str, dict[str,str]]: if VALIDATION_CACHE.exists() and VALIDATION_CACHE.stat().st_size > 1000: data=json.loads(VALIDATION_CACHE.read_text()) else: url='https://api.coingecko.com/api/v3/coins/list?include_platform=false' with urllib.request.urlopen(url, timeout=30) as r: data=json.loads(r.read().decode('utf-8')) VALIDATION_CACHE.write_text(json.dumps(data), encoding='utf-8') return {str(x.get('id','')).lower(): {'symbol':str(x.get('symbol','')).upper(), 'name':str(x.get('name',''))} for x in data} def clean(s:Any)->str: return str(s or '').strip() def decimal_text(v:Any)->tuple[str|None,bool,bool]: if v is None or str(v).strip()=='' or str(v).strip()=='-': return None,False,False raw=str(v).strip().replace("'",'').replace(',', '') sci=bool(re.search(r'[eE][+-]?\d+', raw)) try: d=Decimal(raw) except InvalidOperation: return None,False,sci if d == 0: return None,False,sci if d < 0: raise ValueError('negative quantity') return format(d,'f'), True, sci def guess_symbol(label:str, cg_id:str, cg_meta:dict[str,dict[str,str]])->str: meta=cg_meta.get(cg_id.lower()) if meta: return meta['symbol'].upper() toks=re.findall(r'[A-Za-z0-9]+', label.upper()) return toks[-1] if toks else label[:12].upper() def sem_ok(label:str, cg_id:str, meta:dict[str,str]) -> bool: label_l=label.lower() name_l=meta['name'].lower() sym_l=meta['symbol'].lower() id_l=cg_id.lower().replace('-', ' ') toks=set(re.findall(r'[a-z0-9]+', label_l)) if sym_l and sym_l in toks: return True if name_l and any(t and len(t)>=3 and t in label_l for t in re.findall(r'[a-z0-9]+', name_l)): return True if any(t and len(t)>=3 and t in label_l for t in id_l.split()): return True return False def parse_workbook(cg_meta:dict[str,dict[str,str]]) -> tuple[Plan, list[ParsedHolding], dict[str, Any]]: wb_f=load_workbook(XLSX, data_only=False, read_only=True) wb_v=load_workbook(XLSX, data_only=True, read_only=True) ws_f=wb_f.active; ws_v=wb_v.active rows_f=list(ws_f.iter_rows(values_only=True)); rows_v=list(ws_v.iter_rows(values_only=True)) p=Plan() headers=[clean(x) for x in rows_f[1]] structure={ 'sheet_count': len(wb_f.sheetnames), 'sheet_name': ws_f.title, 'header_api_id': headers[0].lower()=='api id', 'header_coin_name': headers[1].lower()=='coin name', 'wallet_columns_detected': sum(1 for c in WALLET_COLS if c-1 < len(headers) and headers[c-1]), 'rows_nonempty': sum(1 for r in rows_f if any(clean(x) for x in r)), } holdings=[] seen_ids=set() for idx,(rf,rv) in enumerate(zip(rows_f, rows_v), start=1): if idx<=2: continue cg=clean(rf[0] if len(rf)>0 else '').lower() label=clean(rf[1] if len(rf)>1 else '') if not cg and not label: continue p.coin_rows_read += 1 if cg: p.coingecko_id_rows_detected += 1; seen_ids.add(cg) meta=cg_meta.get(cg) if not cg or not meta: p.invalid_or_missing_ids += 1 continue if not sem_ok(label,cg,meta): p.semantic_conflicts += 1 continue p.valid_ids += 1 symbol=guess_symbol(label,cg,cg_meta) for c in WALLET_COLS: wallet=headers[c-1] formula = len(rf)>=c and isinstance(rf[c-1], str) and rf[c-1].startswith('=') if formula: p.formula_cases += 1 try: qty, ok, sci = decimal_text(rv[c-1] if len(rv)>=c else None) except Exception: p.decimal_problems += 1; continue if sci: p.scientific_notation_cases += 1 if not ok or qty is None: continue holdings.append(ParsedHolding(idx,cg,label,symbol,wallet,qty,formula,sci)) p.parsed_holdings=len(holdings) return p, holdings, structure def wallet_type(name:str)->str: n=name.lower() if any(x in n for x in ['binance','kucoin','swissborg']): return 'Exchange' return 'Software Wallet' def build_plan(conn, p:Plan, holdings:list[ParsedHolding]) -> tuple[Plan, list[dict[str,Any]]]: assets_by_cg={clean(r['coingecko_id']).lower(): dict(r) for r in conn.execute('select * from crypto_assets').fetchall() if clean(r['coingecko_id'])} assets_by_symbol={clean(r['symbol']).upper(): dict(r) for r in conn.execute('select * from crypto_assets').fetchall()} wallets={clean(r['wallet_name']).lower(): dict(r) for r in conn.execute('select * from crypto_wallets').fetchall()} existing_holdings={(r['asset_id'], r['wallet_id']) for r in conn.execute('select asset_id,wallet_id from crypto_holdings').fetchall()} asset_plan={}; actions=[]; new_asset_ids=set(); existing_asset_ids=set() for h in holdings: status='skipped'; asset=None; conflict=False if h.coingecko_id in assets_by_cg: asset=assets_by_cg[h.coingecko_id]; status='asset_existing'; existing_asset_ids.add(asset['asset_id']) elif h.symbol_guess.upper() in assets_by_symbol and clean(assets_by_symbol[h.symbol_guess.upper()].get('coingecko_id')).lower() not in ('', h.coingecko_id): p.mapping_conflicts += 1; p.conflicts += 1; p.skipped += 1; conflict=True elif h.symbol_guess.upper() in assets_by_symbol and not clean(assets_by_symbol[h.symbol_guess.upper()].get('coingecko_id')): # conservative: do not update missing mapping silently p.mapping_conflicts += 1; p.conflicts += 1; p.skipped += 1; conflict=True else: aid=stable_id('cryptoasset', h.coingecko_id, h.symbol_guess, '') asset={'asset_id':aid, 'coin_name':h.coin_label, 'symbol':h.symbol_guess, 'coingecko_id':h.coingecko_id} status='asset_new'; new_asset_ids.add(aid) if conflict or not asset: continue wallet_key=h.wallet_label.lower() wallet=wallets.get(wallet_key) wallet_id=wallet['wallet_id'] if wallet else stable_id('wallet', h.wallet_label) if not wallet: p.wallet_missing_importable += 1 exists=(asset['asset_id'], wallet_id) in existing_holdings if exists: p.existing_holdings += 1; p.existing_skipped += 1; hold_status='holding_existing' else: p.new_holdings_importable += 1; hold_status='holding_new' actions.append({'holding':h, 'asset':asset, 'asset_status':status, 'wallet_id':wallet_id, 'wallet_exists': bool(wallet), 'holding_status': hold_status}) p.existing_assets=len(existing_asset_ids) p.new_assets_importable=len(new_asset_ids) p.blockers=p.invalid_or_missing_ids+p.semantic_conflicts+p.mapping_conflicts+p.decimal_problems # block only technical blockers, not skipped conflicts. Import clear subset if at least one new holding and decimal ok. p.import_possible = p.decimal_problems == 0 and p.new_holdings_importable > 0 return p, actions def do_import(conn, p:Plan, actions:list[dict[str,Any]]) -> Plan: before_assets=conn.execute('select count(*) c from crypto_assets').fetchone()['c'] before_holdings=conn.execute('select count(*) c from crypto_holdings').fetchone()['c'] created_assets=set(); created_wallets=set(); errors=[]; imported=0 session_id=create_import_session(conn, import_type='crypto_adjusted_xlsx_initial_snapshot', source_filename=XLSX.name, file_hash=source_hash(XLSX), status='started', rows_total=p.parsed_holdings, rows_imported=0, rows_failed=0, errors=[], notes='Adjusted XLSX with CoinGecko IDs; aggregate-only runtime import; legacy CHF ignored.') p.import_session_created=True try: for a in actions: if a['holding_status']!='holding_new': continue h=a['holding']; asset=a['asset'] if not a['wallet_exists'] and a['wallet_id'] not in created_wallets: create_wallet(conn, wallet_name=h.wallet_label, wallet_type=wallet_type(h.wallet_label), notes='Created during adjusted crypto XLSX import') created_wallets.add(a['wallet_id']) if a['asset_status']=='asset_new' and asset['asset_id'] not in created_assets: create_crypto_asset(conn, coin_name=asset['coin_name'], symbol=asset['symbol'], coingecko_id=asset['coingecko_id'], notes='Created during adjusted crypto XLSX import') created_assets.add(asset['asset_id']) # recheck duplicate immediately before write exists=conn.execute('select 1 from crypto_holdings where asset_id=? and wallet_id=?', (asset['asset_id'], a['wallet_id'])).fetchone() if exists: p.existing_skipped += 1; continue create_initial_holding_snapshot(conn, asset_id=asset['asset_id'], wallet_id=a['wallet_id'], quantity=Decimal(h.quantity_text), verification_status='verified', last_verified_at=SNAPSHOT_DATE+'T00:00:00Z', legacy_snapshot_value_original=None, legacy_snapshot_value_chf=None, legacy_snapshot_currency=None, legacy_snapshot_date=SNAPSHOT_DATE, note='Adjusted crypto XLSX initial snapshot import; CoinGecko ID supplied in source; legacy CHF values ignored.') imported += 1 after_assets=conn.execute('select count(*) c from crypto_assets').fetchone()['c'] after_holdings=conn.execute('select count(*) c from crypto_holdings').fetchone()['c'] p.assets_created=after_assets-before_assets p.holdings_created=after_holdings-before_holdings conn.execute("update import_sessions set status='committed', finished_at=?, rows_imported=?, rows_failed=?, errors_json=?, notes=? where import_session_id=?", (utc_now(), imported, 0, json.dumps(errors), 'Committed adjusted crypto XLSX clear subset; no overwrites; legacy CHF ignored.', session_id)) conn.commit() except Exception as e: errors.append(type(e).__name__) conn.execute("update import_sessions set status='failed', finished_at=?, rows_imported=?, rows_failed=?, errors_json=? where import_session_id=?", (utc_now(), imported, 1, json.dumps(errors), session_id)) conn.commit(); raise # aggregate checks p.audit_events_created=conn.execute("select count(*) c from audit_log where action in ('create_crypto_asset','create_wallet','initial_crypto_holding_snapshot') and created_at >= (select started_at from import_sessions where import_session_id=?)", (session_id,)).fetchone()['c'] new_ids=[x for x in created_assets] if new_ids: q=','.join('?' for _ in new_ids) p.dashboard_crypto_contains_new_coins=conn.execute(f"select count(*) c from crypto_assets where asset_id in ({q})", new_ids).fetchone()['c'] == len(new_ids) p.report_context_contains_new_coins=p.dashboard_crypto_contains_new_coins else: p.dashboard_crypto_contains_new_coins=True; p.report_context_contains_new_coins=True return p def main(): ap=argparse.ArgumentParser(); ap.add_argument('--commit', action='store_true'); args=ap.parse_args() cg=load_cg() p, holdings, structure = parse_workbook(cg) settings=load_settings(repo_root=find_repo_root()) conn=connect(settings.db_path); apply_migrations(conn) p, actions=build_plan(conn,p,holdings) if args.commit: if not p.import_possible: raise SystemExit('DRY_RUN_NOT_CLEAN_FOR_IMPORT') p=do_import(conn,p,actions) print(json.dumps({'mode':'commit' if args.commit else 'dry_run', 'summary':asdict(p), 'structure':structure}, ensure_ascii=False, sort_keys=True)) if __name__=='__main__': main()