10 from typing import Any 11 from zoneinfo import ZoneInfo 12 13 from jarvis_finance.audit.log import record_audit_event 14 from jarvis_finance.crypto.current_balances import current_crypto_balance_basis 15 from jarvis_finance.imports.common import stable_id, utc_now 16 from jarvis_finance.market.providers import MarketDataProvider, PriceQuote 17 from jarvis_finance.services.portfolio_analytics import _exclusive_lock 18 19 SOURCE_KEY = "daily_crypto_current_valuation_v1" 20 ACTIVATION_SOURCE = "market_source_activation_v1" 21 ACTIVATION_ACTION = "market_source_activation_confirmed" 22 SCOPE_KIND = "account" 23 # A virtual aggregate scope; account_id remains NULL because Sprint 20E is not 24 # authorized to create or reclassify financial accounts/performance coverage. 25 SCOPE_ID = "crypto-recorded-holdings" 26 ZERO = Decimal(0) 27 MAX_PROVIDER_AGE_SECONDS = 14_400 # Daily valuation: reject quotes older than four hours. 28 29 30 @dataclass(frozen=True) 31 class CryptoMarketRunResult: 32 run_id: str 33 status: str 34 as_of: str 35 priced_assets: int 36 missing_assets: int 37 price_stored: int 38 valuation_stored: int 39 total_value_chf: str | None 40 reason_codes: tuple[str, ...] 41 idempotent: bool = False 42 43 44 def _fmt(value: Decimal) -> str: 45 return format(value, "f") 46 47 48 def _parse_timestamp(value: str | None) -> datetime | None: 49 if not value: 50 return None 51 try: 52 parsed = datetime.fromisoformat(value) 53 except ValueError: 54 return None 55 if parsed.tzinfo is None: 56 parsed = parsed.replace(tzinfo=UTC) 57 return parsed.astimezone(UTC) 58 59 def _current_inventory(conn: Connection) -> tuple[list[dict[str, Any]], list[str]]: 60 basis = current_crypto_balance_basis(conn) 61 aggregated: dict[str, dict[str, Any]] = {} 62 for (_wallet_id, asset_id), quantity in basis.quantities.items(): 63 row = conn.execute("SELECT symbol,coin_name,coingecko_id FROM crypto_assets WHERE asset_id=? AND is_active=1", (asset_id,)).fetchone() 64 if not row: 65 continue 66 verification = conn.execute("SELECT verification_status FROM crypto_holdings WHERE asset_id=?", (asset_id,)).fetchall() 67 item = aggregated.setdefault( 68 asset_id, 69 { 70 "asset_id": asset_id, 71 "symbol": str(row["symbol"] or ""), 72 "name": str(row["coin_name"] or ""), 73 "provider_id": str(row["coingecko_id"] or ""), 74 "quantity_decimal": ZERO, 75 "all_verified": True, 76 }, 77 ) 78 item["quantity_decimal"] += quantity 79 item["all_verified"] = bool(item["all_verified"] and (basis.confirmed_current or (verification and all(candidate["verification_status"] == "verified" for candidate in verification)))) 80 inventory = [ 81 { 82 "asset_id": item["asset_id"], 83 "symbol": item["symbol"], 84 "name": item["name"], 85 "provider_id": item["provider_id"], 86 "quantity": _fmt(item["quantity_decimal"]), 87 "all_verified": item["all_verified"], 88 } 89 for item in aggregated.values() 90 if item["quantity_decimal"] != ZERO 91 ] 92 reasons: list[str] = [] 93 if not inventory: 94 reasons.append("crypto_recorded_holdings_missing") 95 if any(not row["all_verified"] for row in inventory): 96 reasons.append("crypto_holding_unverified") 97 provider_ids = [row["provider_id"] for row in inventory if row["provider_id"]] 98 if len(provider_ids) != len(inventory): 99 reasons.append("crypto_provider_mapping_missing") 100 if len(set(provider_ids)) != len(provider_ids): 101 reasons.append("crypto_provider_mapping_ambiguous") 102 return inventory, sorted(set(reasons)) 103 104 105 def _provider_quotes( 106 provider: MarketDataProvider, 107 provider_ids: list[str], 108 currency: str, 109 ) -> dict[str, PriceQuote]: 110 if hasattr(provider, "get_crypto_prices"): 111 return provider.get_crypto_prices(provider_ids, currency) # type: ignore[attr-defined] 112 return {provider_id: provider.get_crypto_price(provider_id, currency) for provider_id in provider_ids} 113 114 115 def _stable_input_fingerprint(*, provider: str, currency: str, items: list[dict[str, Any]]) -> str: 116 stable_items = [ 117 { 118 "asset_id": item["asset_id"], 119 "provider_id": item["provider_id"], 120 "quantity": item["quantity"], 121 "currency": item["currency"], 122 "price": item["price"], 123 "provider": item["provider"], 124 "provider_timestamp": item["provider_timestamp"], 125 "quality_status": item["quality_status"], 126 } 127 for item in items 128 ] 129 payload = {"provider": provider, "currency": currency, "items": stable_items} 130 return hashlib.sha256( 131 json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() 132 ).hexdigest() 133 134 135 def build_crypto_market_dry_run( 136 conn: Connection, 137 *, 138 provider: MarketDataProvider, 139 now: datetime | None = None, 140 currency: str = "CHF", 141 max_provider_age_seconds: int = MAX_PROVIDER_AGE_SECONDS, 142 ) -> dict[str, Any]: 143 """Fetch and reconcile current crypto quotes without writing to SQLite.""" 144 145 current = now or datetime.now(UTC) 146 if current.tzinfo is None: 147 current = current.replace(tzinfo=UTC) 148 currency = currency.upper() 149 inventory, reasons = _current_inventory(conn) 150 if getattr(provider, "provider_key", None) != "coingecko": 151 reasons.append("crypto_provider_not_approved") 152 provider_ids = [str(row["provider_id"]) for row in inventory if row["provider_id"]] 153 quotes: dict[str, PriceQuote] = {} 154 if provider_ids and not reasons: 155 try: 156 quotes = _provider_quotes(provider, provider_ids, currency) 157 except Exception as exc: # noqa: BLE001 - provider boundary becomes a classified dry-run result 158 reasons.append("crypto_provider_not_reachable") 168 } 169 170 items: list[dict[str, Any]] = [] 171 total = ZERO 172 for row in inventory: 173 quote = quotes.get(str(row["provider_id"])) 174 provider_ts = _parse_timestamp(quote.provider_timestamp if quote else None) 175 age_seconds = int((current - provider_ts).total_seconds()) if provider_ts else None 176 item_reasons: list[str] = [] 177 if not row["provider_id"]: 178 item_reasons.append("crypto_provider_mapping_missing") 179 elif quote is not None and quote.coingecko_id != row["provider_id"]: 180 item_reasons.append("crypto_provider_identity_mismatch") 181 if quote is not None and quote.currency.upper() != currency: 182 item_reasons.append("crypto_provider_currency_mismatch") 183 if quote is not None and quote.provider != "CoinGecko": 184 item_reasons.append("crypto_provider_not_approved") 185 if row["provider_id"] and (quote is None or quote.price is None): 186 item_reasons.append("crypto_price_missing") 187 if quote is not None and quote.quality_status != "fresh": 188 item_reasons.append("crypto_price_not_fresh") 189 if quote is not None and quote.price is not None and quote.price <= ZERO: 190 item_reasons.append("crypto_price_invalid") 191 if provider_ts is None: 192 item_reasons.append("crypto_provider_timestamp_missing") 193 elif age_seconds is not None and age_seconds < -300: 194 item_reasons.append("crypto_provider_timestamp_future") 195 elif age_seconds is not None and age_seconds > max_provider_age_seconds: 196 item_reasons.append("crypto_provider_timestamp_stale") 197 value = None 198 if not item_reasons and quote and quote.price is not None: 199 value = Decimal(str(row["quantity"])) * quote.price 200 total += value 201 items.append( 202 { 203 "asset_id": row["asset_id"], 204 "symbol": row["symbol"], 205 "name": row["name"], 206 "provider_id": row["provider_id"] or None, 207 "quantity": row["quantity"], 208 "currency": quote.currency.upper() if quote else currency, 209 "price": _fmt(quote.price) if quote and quote.price is not None else None, 210 "value_chf": _fmt(value) if value is not None else None, 211 "provider": quote.provider if quote else None, 212 "provider_timestamp": quote.provider_timestamp if quote else None, 213 "provider_age_seconds": age_seconds, 214 "quality_status": quote.quality_status if quote else "missing", 215 "reason_codes": sorted(set(item_reasons)), 216 } 217 ) 218 item_reasons = [reason for item in items for reason in item["reason_codes"]] 219 reasons = sorted({*reasons, *item_reasons}) 220 priced = sum(not item["reason_codes"] for item in items) 221 if not items or priced == 0: 222 status = "blocked" 223 elif priced != len(items) or reasons: 224 status = "partial" 225 else: 226 status = "complete" 227 payload = { 228 "status": status, 229 "provider": "CoinGecko", 230 "currency": currency, 231 "requested_at": current.isoformat(), 232 "asset_count": len(items), 233 "priced_count": priced, 234 "missing_count": len(items) - priced, 235 "total_value_chf": _fmt(total) if status == "complete" else None, 236 "items": items, 237 "reason_codes": reasons, 238 "planned_writes": { 239 "crypto_prices": len(items) if status == "complete" else 0, 240 "fx_rates": 0, 241 "portfolio_valuation_snapshots": 1 if status == "complete" else 0, 242 "market_data_runs": 1, 243 "audit_log": 1, 244 }, 245 "persistence_performed": False, 246 } 247 payload["input_fingerprint"] = _stable_input_fingerprint( 248 provider=payload["provider"], currency=payload["currency"], items=items 249 ) 250 return payload 251 252 253 def is_crypto_market_source_activated(conn: Connection) -> bool: 254 return bool( 255 conn.execute( 256 """SELECT 1 FROM audit_log 257 WHERE source=? AND action=? 258 AND json_extract(new_values_json,'$.source')='crypto' 259 AND json_extract(new_values_json,'$.enabled')=1 260 LIMIT 1""", 261 (ACTIVATION_SOURCE, ACTIVATION_ACTION), 262 ).fetchone() 263 ) 264 265 266 def activate_crypto_market_source( 267 conn: Connection, 268 *, 269 confirmation_id: str, 270 note: str = "Controlled crypto market-data source activation", 271 ) -> dict[str, Any]: 272 confirmation_id = confirmation_id.strip() 273 if not confirmation_id: 274 raise ValueError("confirmation_id is required") 275 conn.execute("BEGIN IMMEDIATE") 276 try: 277 prior = conn.execute( 278 "SELECT audit_id FROM audit_log WHERE source=? AND action=? AND entity_id=? LIMIT 1", 279 (ACTIVATION_SOURCE, ACTIVATION_ACTION, confirmation_id), 280 ).fetchone() 281 if prior: 282 conn.commit() 283 return { 284 "source": "crypto", 285 "enabled": True, 286 "idempotent": True, 287 "audit_id": str(prior["audit_id"]), 288 } 289 audit_id = record_audit_event( 290 conn, 291 source=ACTIVATION_SOURCE, 292 action=ACTIVATION_ACTION, 293 entity_type="market_source", 295 old_values={"enabled": False}, 296 new_values={"source": "crypto", "enabled": True}, 297 user_text_note=note, 298 created_by="system", 299 ) 300 conn.commit() 301 except Exception: 302 conn.rollback() 303 raise 304 return {"source": "crypto", "enabled": True, "idempotent": False, "audit_id": audit_id} 305 306 307 def _existing_complete(conn: Connection, day: str) -> Any: 308 return conn.execute( 309 """SELECT run_id,as_of,status,started_at,input_fingerprint, 310 price_total,price_stored,valuation_stored,reason_codes_json 311 FROM market_data_runs WHERE source_key=? AND as_of=? AND status='complete' 312 ORDER BY completed_at DESC,run_id DESC LIMIT 1""", 313 (SOURCE_KEY, day), 314 ).fetchone() 315 316 317 def _assert_existing_complete_matches_current_inputs(conn: Connection, row: Any) -> None: 318 inventory, reasons = _current_inventory(conn) 319 if reasons: 320 raise RuntimeError("crypto_existing_run_input_drift") 321 price_rows = conn.execute( 322 """SELECT asset_id,coingecko_id,price_currency,price,provider, 323 provider_timestamp,quality_status 324 FROM crypto_prices 325 WHERE fetched_at=? AND provider='CoinGecko' 326 ORDER BY asset_id,crypto_price_id""", 327 (row["started_at"],), 328 ).fetchall() 329 if len(price_rows) != len(inventory): 330 raise RuntimeError("crypto_existing_run_input_drift") 331 by_asset = {str(price_row["asset_id"]): price_row for price_row in price_rows} 332 if len(by_asset) != len(price_rows): 333 raise RuntimeError("crypto_existing_run_input_drift") 334 items: list[dict[str, Any]] = [] 335 for asset in inventory: 336 price_row = by_asset.get(str(asset["asset_id"])) 337 if not price_row or str(price_row["coingecko_id"]) != str(asset["provider_id"]): 338 raise RuntimeError("crypto_existing_run_input_drift") 339 items.append( 340 { 341 "asset_id": asset["asset_id"], 342 "provider_id": asset["provider_id"], 343 "quantity": asset["quantity"], 344 "currency": str(price_row["price_currency"]).upper(), 345 "price": str(price_row["price"]), 346 "provider": str(price_row["provider"]), 347 "provider_timestamp": str(price_row["provider_timestamp"]), 348 "quality_status": str(price_row["quality_status"]), 349 } 350 ) 351 fingerprint = _stable_input_fingerprint( 352 provider="CoinGecko", currency="CHF", items=items 353 ) 354 if fingerprint != str(row["input_fingerprint"]): 355 raise RuntimeError("crypto_existing_run_input_drift") 356 357 358 def _complete_result(conn: Connection, row: Any, *, day: str) -> CryptoMarketRunResult: 359 valuation = conn.execute( 360 """SELECT value_original FROM portfolio_valuation_snapshots 361 WHERE scope_kind=? AND scope_id=? AND source=? AND substr(valuation_at,1,10)=? 362 ORDER BY captured_at DESC,snapshot_id DESC LIMIT 1""", 363 (SCOPE_KIND, SCOPE_ID, SOURCE_KEY, day), 364 ).fetchone() 365 return CryptoMarketRunResult( 366 run_id=str(row["run_id"]), 367 status="complete", 368 as_of=day, 369 priced_assets=int(row["price_total"] or 0), 370 missing_assets=0, 371 price_stored=0, 372 valuation_stored=0, 373 total_value_chf=str(valuation["value_original"]) if valuation else None, 374 reason_codes=tuple(json.loads(row["reason_codes_json"] or "[]")), 375 idempotent=True, 376 ) 377 378 379 def _persist_unsuccessful_attempt(conn: Connection, *, day: str, preview: dict[str, Any]) -> str: 380 fingerprint = str(preview["input_fingerprint"]) 381 run_id = stable_id("market-run-attempt", SOURCE_KEY, day, fingerprint) 382 prior = conn.execute("SELECT run_id FROM market_data_runs WHERE run_id=?", (run_id,)).fetchone() 383 if prior: 384 return str(prior["run_id"]) 385 now = utc_now() 386 missing = [ 387 { 388 "asset_id": item["asset_id"], 389 "provider_id": item["provider_id"], 390 "reason_codes": item["reason_codes"], 391 } 392 for item in preview["items"] 393 if item["reason_codes"] 394 ] 395 conn.execute("BEGIN IMMEDIATE") 396 try: 397 audit_id = record_audit_event( 398 conn, 399 source=SOURCE_KEY, 400 action="crypto_market_one_shot_incomplete", 401 entity_type="market_data_run", 402 entity_id=run_id, 403 old_values={}, 404 new_values={ 405 "as_of": day, 406 "status": preview["status"], 407 "provider": preview["provider"], 408 "priced_assets": preview["priced_count"], 409 "missing_assets": preview["missing_count"], 410 "writes": 0, 411 "reason_codes": preview["reason_codes"], 412 "input_fingerprint": fingerprint, 413 }, 414 created_by="system", 415 ) 416 conn.execute( 417 """INSERT INTO market_data_runs( 418 run_id,source_key,as_of,input_fingerprint,status,started_at,completed_at, 419 price_total,price_stored,fx_total,fx_stored,benchmark_total,benchmark_stored, 420 valuation_stored,missing_instruments_json,reason_codes_json,audit_id 421 ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", 422 ( 423 run_id, 424 SOURCE_KEY, 425 day, 426 fingerprint, 427 preview["status"], 428 now, 429 utc_now(), 430 preview["asset_count"], 431 0, 432 0, 433 0, 434 0, 435 0, 436 0, 437 json.dumps(missing, sort_keys=True), 438 json.dumps(preview["reason_codes"], sort_keys=True), 439 audit_id, 440 ), 441 ) 442 conn.commit() 443 except Exception: 444 conn.rollback() 445 raise 446 return run_id 447 448 449 def run_crypto_market_one_shot( 450 conn: Connection, 451 *, 452 provider: MarketDataProvider, 453 as_of: str | None = None, 454 lock_path: Path | None = None, 455 now: datetime | None = None, 456 require_activation: bool = False, 457 ) -> CryptoMarketRunResult: 458 """Atomically persist one current-day crypto price set and aggregate valuation.""" 459 460 requested = date.fromisoformat(as_of) if as_of else datetime.now(ZoneInfo("Europe/Zurich")).date() 461 today = (now or datetime.now(ZoneInfo("Europe/Zurich"))).astimezone(ZoneInfo("Europe/Zurich")).date() 462 if requested != today: 463 raise ValueError("current crypto market one-shot cannot backdate recorded holdings") 464 day = requested.isoformat() 465 lock = lock_path or Path("/tmp/jarvis-finance-crypto-market.lock") 466 with _exclusive_lock(lock): 467 if require_activation and not is_crypto_market_source_activated(conn): 468 return CryptoMarketRunResult( 469 run_id="", 470 status="not_activated", 471 as_of=day, 472 priced_assets=0, 473 missing_assets=0, 474 price_stored=0, 475 valuation_stored=0, 476 total_value_chf=None, 477 reason_codes=("crypto_market_source_activation_required",), 478 ) 479 existing = _existing_complete(conn, day) 480 if existing: 481 _assert_existing_complete_matches_current_inputs(conn, existing) 482 return _complete_result(conn, existing, day=day) 483 preview = build_crypto_market_dry_run(conn, provider=provider, now=now) 484 if preview["status"] != "complete": 485 run_id = _persist_unsuccessful_attempt(conn, day=day, preview=preview) 486 return CryptoMarketRunResult( 487 run_id=run_id, 488 status=str(preview["status"]), 489 as_of=day, 490 priced_assets=int(preview["priced_count"]), 491 missing_assets=int(preview["missing_count"]), 492 price_stored=0, 493 valuation_stored=0, 494 total_value_chf=None, 495 reason_codes=tuple(preview["reason_codes"]), 496 ) 497 fingerprint = str(preview["input_fingerprint"]) 498 run_id = stable_id("market-run", SOURCE_KEY, day) 499 captured_at = utc_now() 500 snapshot_id = stable_id("crypto-current-valuation", SOURCE_KEY, day) 501 conn.execute("BEGIN IMMEDIATE") 502 try: 503 concurrent_complete = _existing_complete(conn, day) 504 if concurrent_complete: 505 _assert_existing_complete_matches_current_inputs(conn, concurrent_complete) 506 conn.rollback() 507 return _complete_result(conn, concurrent_complete, day=day) 508 inventory, current_reasons = _current_inventory(conn) 509 inventory_projection = [ 510 (row["asset_id"], row["provider_id"], row["quantity"]) 511 for row in inventory 512 ] 513 preview_projection = [ 514 (row["asset_id"], row["provider_id"], row["quantity"]) 515 for row in preview["items"] 516 ] 517 if current_reasons or inventory_projection != preview_projection: 518 raise RuntimeError("crypto_inventory_changed_after_provider_fetch") 519 for item in preview["items"]: 520 price_id = stable_id( 521 "cryptoprice", 522 item["asset_id"], 523 item["provider_id"], 524 "CHF", 525 item["provider"], 526 item["provider_timestamp"], 527 ) 528 conn.execute( 529 """INSERT INTO crypto_prices( 530 crypto_price_id,asset_id,coingecko_id,price_currency,price,provider, 531 provider_timestamp,fetched_at,quality_status,error_message 532 ) VALUES(?,?,?,?,?,?,?,?,?,NULL)""", 533 ( 534 price_id, 535 item["asset_id"], 536 item["provider_id"], 537 "CHF", 538 item["price"], 539 item["provider"], 540 item["provider_timestamp"], 541 captured_at, 542 "fresh", 543 ), 544 ) 545 conn.execute( 546 """INSERT INTO portfolio_valuation_snapshots( 547 snapshot_id,scope_kind,scope_id,account_id,value_original,currency, 548 base_currency,fx_rate_to_base,fx_direction,valuation_at,source,captured_at, 549 snapshot_version,supersedes_snapshot_id,source_reference,quality_status,reason_codes_json 550 ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", 551 ( 552 snapshot_id, 553 SCOPE_KIND, 554 SCOPE_ID, 555 None, 556 preview["total_value_chf"], 578 new_values={ 579 "as_of": day, 580 "status": "complete", 581 "provider": preview["provider"], 582 "priced_assets": preview["priced_count"], 583 "missing_assets": 0, 584 "price_rows_written": len(preview["items"]), 585 "valuation_rows_written": 1, 586 "input_fingerprint": fingerprint, 587 }, 588 created_by="system", 589 ) 590 conn.execute( 591 """INSERT INTO market_data_runs( 592 run_id,source_key,as_of,input_fingerprint,status,started_at,completed_at, 593 price_total,price_stored,fx_total,fx_stored,benchmark_total,benchmark_stored, 594 valuation_stored,missing_instruments_json,reason_codes_json,audit_id 595 ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", 596 ( 597 run_id, 598 SOURCE_KEY, 599 day, 600 fingerprint, 601 "complete", 602 captured_at, 603 utc_now(), 604 len(preview["items"]), 605 len(preview["items"]), 606 0, 607 0, 608 0, 609 0, 610 1, 611 "[]", 612 "[]", 613 audit_id, 614 ), 615 ) 616 conn.commit() 617 except Exception: 618 conn.rollback() 619 raise 620 return CryptoMarketRunResult( 621 run_id=run_id, 622 status="complete", 623 as_of=day, 624 priced_assets=len(preview["items"]), 625 missing_assets=0, 626 price_stored=len(preview["items"]), 627 valuation_stored=1, 628 total_value_chf=str(preview["total_value_chf"]), 629 reason_codes=(), 630 idempotent=False, 631 ) __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23__HERMES_CWD_8d46a20096ed__