1 from __future__ import annotations 2 3 from dataclasses import replace 4 from concurrent.futures import ThreadPoolExecutor, as_completed 5 from datetime import date, datetime, timedelta, timezone 6 from decimal import Decimal, InvalidOperation 7 import json 8 import time 9 from sqlite3 import Connection 10 from typing import Any 11 from urllib import error, parse, request 12 13 from fastapi import HTTPException 14 15 from jarvis_finance.api.schemas.market import ChartPoint, EquityCandlesResponse, MarketBatchUpdateResponse, MarketChartResponse, MarketQuoteResponse, MarketStatusResponse, QuoteRefreshRequest 16 from jarvis_finance.imports.common import utc_now 17 from jarvis_finance.market.providers import CoinGeckoClient, PriceQuote, store_crypto_price 18 from jarvis_finance.market_data.cache import get_crypto_chart_points, get_equity_chart_points, get_equity_intraday_candles, upsert_crypto_price_point, upsert_equity_intraday_candles, upsert_equity_price_point 19 from jarvis_finance.market_data.prices import EquityPriceQuote, equity_price_provider_by_name, exchange_matches, provider_capability, store_market_price 20 from jarvis_finance.storage.database import connect 21 22 23 def _decimal_text(value: object | None) -> str | None: 24 if value in (None, ""): 25 return None 26 try: 27 return format(Decimal(str(value)), "f") 28 except (InvalidOperation, ValueError): 29 return None 30 31 32 def _quality_from_error(message: str | None, default: str = "missing") -> str: 33 msg = (message or "").lower() 34 if "rate_limited" in msg or "429" in msg: 35 return "rate_limited" 36 if "endpoint_restricted" in msg or "plan_restricted" in msg or "402" in msg: 37 return "plan_restricted" 38 if "auth_failed" in msg or "auth_error" in msg or "401" in msg or "403" in msg: 39 return "auth_error" 40 if "network_error" in msg: 41 return "network_error" 42 if "api_key_missing" in msg or "provider_symbol_missing" in msg or "missing" in msg: 43 return "missing" 44 if "unsupported" in msg: 45 return "unsupported_pair" 46 if "stale" in msg: 47 return "stale" 48 return default 49 50 51 def _chart_response(points, *, currency: str, provider_symbol: str | None = None, warnings: list[str] | None = None) -> MarketChartResponse: 52 chart_points = [ChartPoint(timestamp=str(r["timestamp"]), price=str(r["price"]), currency=str(r["currency"] or currency), provider=r["provider"], quality_status=r["source_quality"]) for r in points] 53 latest = chart_points[-1] if chart_points else None 54 first = chart_points[0] if chart_points else None 55 change_abs = None 56 change_pct = None 57 if latest and first: 58 try: 59 start = Decimal(first.price) 60 end = Decimal(latest.price) 61 change_abs = format(end - start, "f") 62 change_pct = format(((end - start) / start * Decimal("100")), "f") if start else None 63 except Exception: 64 pass 65 return MarketChartResponse( 66 latest_price=latest.price if latest else None, 67 currency=latest.currency if latest else currency, 68 change_abs=change_abs, 69 change_pct=change_pct, 70 close=latest.price if latest else None, 71 provider=latest.provider if latest else None, 72 provider_symbol=provider_symbol, 73 fetched_at=latest.timestamp if latest else None, 74 quality_status=latest.quality_status if latest and latest.quality_status else ("fresh" if latest else "missing"), 75 chart_points=chart_points, 76 warnings=warnings or ([] if chart_points else ["Noch zu wenig Kursdaten"]), 77 ) 78 79 80 def get_market_status(conn: Connection) -> MarketStatusResponse: 81 def scalar(sql: str, params: tuple = ()): 82 row = conn.execute(sql, params).fetchone() 83 return row[0] if row else None 84 return MarketStatusResponse( 85 equity_latest_update=scalar("SELECT MAX(fetched_at) FROM equity_price_points"), 86 crypto_latest_update=scalar("SELECT MAX(fetched_at) FROM crypto_price_points"), 87 equity_cached_points=int(scalar("SELECT COUNT(*) FROM equity_price_points") or 0), 88 crypto_cached_points=int(scalar("SELECT COUNT(*) FROM crypto_price_points") or 0), 89 mapped_equity_instruments=int(scalar("SELECT COUNT(DISTINCT instrument_id) FROM instrument_price_mappings WHERE mapping_status='mapped' AND provider_symbol IS NOT NULL") or 0), 90 mapped_crypto_assets=int(scalar("SELECT COUNT(*) FROM crypto_assets WHERE is_active=1 AND coingecko_id IS NOT NULL AND coingecko_id!=''") or 0), 91 render_provider_calls=False, 92 warnings=[], 93 ) 94 95 96 def _instrument_mapping(conn: Connection, instrument_id: str): 97 inst = conn.execute("SELECT instrument_id, provider_symbol, data_provider_primary, exchange, currency, is_active, instrument_status, valuation_policy FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone() 98 if not inst: 99 raise HTTPException(status_code=404, detail="Instrument not found") 100 if not bool(inst["is_active"]) or str(inst["instrument_status"] or "active").lower() in {"inactive", "delisted", "suspended", "merged"}: 101 return inst, None, ["Instrument ist nicht für automatische Kursaktualisierung aktiv"] 102 if str(inst["valuation_policy"] or "").lower() == "exclude_from_auto_price_update": 103 return inst, None, ["Instrument ist durch die Bewertungsrichtlinie von automatischen Kursaktualisierungen ausgeschlossen"] 104 mapping = conn.execute("SELECT * FROM instrument_price_mappings WHERE instrument_id=? AND mapping_status='mapped' AND provider_symbol IS NOT NULL ORDER BY CASE provider WHEN 'fmp' THEN 1 WHEN 'finnhub' THEN 2 WHEN 'twelvedata' THEN 3 ELSE 4 END LIMIT 1", (instrument_id,)).fetchone() 105 provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None 106 if not provider_symbol: 107 return inst, None, ["Provider-Symbol fehlt"] 108 return inst, mapping, [] 109 110 111 def _effective_market_date(value: str | None = None) -> date: 112 result = date.fromisoformat(value) if value else datetime.now(timezone.utc).date() 113 while result.weekday() >= 5: 114 result -= timedelta(days=1) 115 return result 116 117 118 def _business_day_age(earlier: date, later: date) -> int: 119 if earlier > later: 120 return -1 121 cursor = earlier 122 age = 0 123 while cursor < later: 124 cursor += timedelta(days=1) 125 if cursor.weekday() < 5: 126 age += 1 127 return age 128 129 130 def _has_fresh_price_for_target( 131 conn: Connection, 132 instrument_id: str, 133 target: date, 134 *, 135 stale_before: str | None = None, 136 ) -> bool: 137 mapping = conn.execute( 138 """SELECT provider,provider_symbol,provider_market,upper(COALESCE(trading_currency,currency,'')) currency 139 FROM instrument_price_mappings 140 WHERE instrument_id=? AND mapping_status='mapped' AND provider_symbol IS NOT NULL 141 ORDER BY CASE provider WHEN 'fmp' THEN 1 ELSE 2 END,updated_at DESC LIMIT 1""", 142 (instrument_id,), 143 ).fetchone() 144 if not mapping: 145 return False 146 rows = conn.execute( 147 """SELECT price_date,currency,provider,provider_symbol,provider_market, 148 COALESCE(fetched_at,created_at,price_timestamp,price_date) freshness_at 149 FROM market_prices 150 WHERE instrument_id=? AND price_date<=? AND close IS NOT NULL AND close!='' 151 AND quality_status='fresh' AND error_message IS NULL 152 ORDER BY price_date DESC,COALESCE(fetched_at,created_at) DESC""", 153 (instrument_id, target.isoformat()), 154 ).fetchall() 155 for row in rows: 156 if stale_before and str(row["freshness_at"] or "") < stale_before: 157 continue 158 actual_date = date.fromisoformat(str(row["price_date"])[:10]) 159 if not 0 <= _business_day_age(actual_date, target) <= 2: 160 continue 161 if str(row["provider_symbol"] or "").upper() != str(mapping["provider_symbol"] or "").upper(): 162 continue 163 if str(row["currency"] or "").upper() != str(mapping["currency"] or "").upper(): 164 continue 165 if not exchange_matches(mapping["provider_market"], row["provider_market"]): 166 continue 167 if str(row["provider"] or "").lower() != str(mapping["provider"] or "").lower(): 168 if str(row["provider"] or "").lower() != "yfinance": 169 continue 170 return True 171 return False 172 173 174 def _quote_date(quote: EquityPriceQuote, fallback: date) -> date: 175 if quote.price_timestamp: 176 try: 177 return datetime.fromisoformat(quote.price_timestamp.replace("Z", "+00:00")).date() 178 except ValueError: 179 try: 180 return date.fromisoformat(quote.price_timestamp[:10]) 181 except ValueError: 182 pass 183 return fallback 184 185 186 def _validate_historical_quote(quote: EquityPriceQuote, *, mapping: Any, target: date, requested_provider: str) -> EquityPriceQuote: 187 if quote.close is None or quote.close <= 0: 188 return replace(quote, quality_status=_quality_from_error(quote.error_message, quote.quality_status)) 189 quote_date = _quote_date(quote, target) 190 if quote_date > target: 191 return replace(quote, close=None, quality_status="future_price_rejected", error_message="future_price_rejected") 192 if _business_day_age(quote_date, target) > 2: 193 return replace(quote, close=None, quality_status="stale", error_message="historical_price_too_old") 194 expected_currency = str(mapping["trading_currency"] or mapping["currency"] or "").upper() 195 actual_currency = str(quote.currency or "").upper() 196 is_fallback = requested_provider == "auto" and str(quote.provider or "").lower() == "yfinance" 197 if is_fallback: 198 capability = provider_capability(str(quote.provider)) 199 if not capability.supports_historical_as_of: 200 return replace(quote, close=None, quality_status="provider_not_historical", error_message="provider_not_historical") 201 if not actual_currency or (expected_currency and actual_currency != expected_currency): 202 return replace(quote, close=None, quality_status="currency_mismatch", error_message="currency_mismatch") 203 if str(quote.provider_symbol or "").upper() != str(mapping["provider_symbol"] or "").upper(): 204 return replace(quote, close=None, quality_status="symbol_mismatch", error_message="symbol_mismatch") 205 if not exchange_matches(str(mapping["provider_market"] or ""), quote.provider_market): 206 return replace(quote, close=None, quality_status="exchange_mismatch", error_message="exchange_mismatch") 207 elif actual_currency and expected_currency and actual_currency != expected_currency: 208 return replace(quote, close=None, quality_status="currency_mismatch", error_message="currency_mismatch") 209 return replace(quote, currency=actual_currency or expected_currency, price_timestamp=quote_date.isoformat(), quality_status="fresh") 210 211 212 def refresh_equity_quote(conn: Connection, instrument_id: str, req: QuoteRefreshRequest) -> MarketQuoteResponse: 213 inst, mapping, warnings = _instrument_mapping(conn, instrument_id) 214 if warnings: 215 return MarketQuoteResponse(provider_symbol=None, currency=inst["currency"] if inst else None, quality_status="missing", warnings=warnings) 216 mapping_data = dict(mapping) if mapping else { 217 "provider": inst["data_provider_primary"] or "auto", 218 "provider_symbol": inst["provider_symbol"], 219 "provider_market": inst["exchange"], 220 "trading_currency": inst["currency"], 221 "currency": inst["currency"], 222 } 223 provider_symbol = str(mapping_data["provider_symbol"]) 224 provider_name = (req.provider or "auto").lower() 225 target = _effective_market_date(req.price_date) 226 quote = equity_price_provider_by_name(provider_name).get_price(provider_symbol, price_date=target.isoformat()) 227 quote = _validate_historical_quote(quote, mapping=mapping_data, target=target, requested_provider=provider_name) 228 quality = quote.quality_status if quote.close is not None else _quality_from_error(quote.error_message, quote.quality_status) 229 ts = quote.price_timestamp or target.isoformat() 230 if not req.dry_run and quote.close is not None and quality == "fresh": 231 store_market_price( 232 conn, 233 instrument_id=instrument_id, 234 price_date=ts[:10], 235 close=quote.close, 236 currency=quote.currency or inst["currency"] or "CHF", 237 provider=quote.provider, 238 provider_symbol=quote.provider_symbol or provider_symbol, 239 provider_market=quote.provider_market or mapping_data["provider_market"], 240 price_timestamp=ts, 241 adjusted_close=quote.adjusted_close, 242 quality_status=quality, 243 error_message=None, 244 ) 245 upsert_equity_price_point( 246 conn, 247 instrument_id=instrument_id, 248 timestamp=ts, 249 price=quote.close, 250 currency=quote.currency or inst["currency"] or "CHF", 251 provider=quote.provider, 252 provider_symbol=quote.provider_symbol or provider_symbol, 253 interval=req.interval, 254 source_quality=quality, 255 ) 256 conn.commit() 257 return MarketQuoteResponse( 258 latest_price=_decimal_text(quote.close), 259 currency=quote.currency or inst["currency"], 260 close=_decimal_text(quote.close), 261 provider=quote.provider, 262 provider_symbol=quote.provider_symbol or provider_symbol, 263 fetched_at=ts, 264 quality_status=quality, 265 warnings=warnings + ([quote.error_message] if quote.error_message else []), 266 ) 267 268 269 def _persistent_database_path(conn: Connection) -> str | None: 270 row = next((row for row in conn.execute("PRAGMA database_list") if str(row[1]) == "main"), None) 271 return str(row[2]) if row and str(row[2] or "") else None 272 273 274 def _parallel_equity_worker( 275 db_path: str, instrument_id: str, req: QuoteRefreshRequest 276 ) -> tuple[MarketQuoteResponse, int]: 277 worker = connect(db_path) 278 worker.execute("PRAGMA busy_timeout=10000") 279 try: 280 attempt = 0 281 quote: MarketQuoteResponse | None = None 282 while attempt <= req.max_retries: 283 attempt += 1 284 quote = refresh_equity_quote(worker, instrument_id, req) 285 if quote.latest_price is not None and quote.quality_status == "fresh": 286 break 287 if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries: 288 break 289 time.sleep(min(2 ** (attempt - 1), 4)) 290 assert quote is not None 291 return quote, attempt 292 finally: 293 worker.close() 294 295 296 def refresh_equity_quotes_batch(conn: Connection, req: QuoteRefreshRequest) -> MarketBatchUpdateResponse: 297 requested_at = utc_now() 298 target_date = _effective_market_date(req.price_date) 299 target = target_date.isoformat() 300 all_rows = conn.execute( 301 """ 302 SELECT DISTINCT i.instrument_id,i.name,i.ticker 303 FROM instruments i 304 JOIN instrument_price_mappings m 305 ON m.instrument_id=i.instrument_id AND m.mapping_status='mapped' 306 WHERE i.asset_class IN ('stock','equity','etf') 307 AND i.is_active=1 308 AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') 309 AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update' 310 AND m.provider_symbol IS NOT NULL AND m.provider_symbol!='' 311 ORDER BY i.name 312 """, 313 ).fetchall() 314 row_states = [ 315 ( 316 row, 317 _has_fresh_price_for_target( 318 conn, 319 str(row["instrument_id"]), 320 target_date, 321 stale_before=req.stale_before, 322 ), 323 ) 324 for row in all_rows 325 ] 326 row_states.sort(key=lambda item: (item[1], str(item[0]["name"] or ""))) 327 bounded_limit = max(1, min(int(req.limit or 100), 500)) 328 row_states = row_states[:bounded_limit] 329 updated = skipped = cached = processed = 0 330 would_update = provider_calls = 0 331 successful_instruments: set[str] = set() 332 result_dates: list[str] = [] 333 warnings: list[str] = [] 334 errors: list[str] = [] 335 item_results: list[dict[str, str | int | bool | None]] = [] 336 last_call_at = 0.0 337 parallel_results: dict[str, tuple[MarketQuoteResponse, int]] = {} 338 db_path = _persistent_database_path(conn) 339 uncached_rows = [row for row, has_fresh in row_states if not (req.only_missing and has_fresh)] 340 if db_path and not req.dry_run and req.max_parallelism > 1 and len(uncached_rows) > 1: 341 with ThreadPoolExecutor(max_workers=min(req.max_parallelism, len(uncached_rows))) as executor: 342 futures = { 343 executor.submit(_parallel_equity_worker, db_path, str(row["instrument_id"]), req): str(row["instrument_id"]) 344 for row in uncached_rows 345 } 346 for future in as_completed(futures): 347 instrument_id = futures[future] 348 try: 349 parallel_results[instrument_id] = future.result() 350 except Exception as exc: 351 parallel_results[instrument_id] = ( 352 MarketQuoteResponse(quality_status="provider_error", warnings=[type(exc).__name__]), 353 1, 354 ) 355 for row, has_fresh in row_states: 356 if req.only_missing and has_fresh: 357 cached += 1 358 skipped += 1 359 item_results.append({"instrument_id": row["instrument_id"], "ticker": row["ticker"], "status": "cached", "attempts": 0}) 360 continue 361 parallel = parallel_results.get(str(row["instrument_id"])) 362 if parallel: 363 quote, attempt = parallel 364 processed += attempt 365 provider_calls += attempt 366 else: 367 elapsed = time.monotonic() - last_call_at 368 if last_call_at and elapsed < req.pacing_seconds: 369 time.sleep(req.pacing_seconds - elapsed) 370 attempt = 0 371 quote: MarketQuoteResponse | None = None 372 while attempt <= req.max_retries: 373 attempt += 1 374 processed += 1 375 provider_calls += 1 376 last_call_at = time.monotonic() 377 quote = refresh_equity_quote(conn, row["instrument_id"], req) 378 if quote.latest_price is not None and quote.quality_status == "fresh": 379 break 380 if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries: 381 break 382 time.sleep(min(2 ** (attempt - 1), 4)) 383 assert quote is not None 384 if quote.latest_price is not None and quote.quality_status == "fresh": 385 successful_instruments.add(str(row["instrument_id"])) 386 if quote.fetched_at: 387 result_dates.append(quote.fetched_at[:10]) 388 if req.dry_run: 389 would_update += 1 390 else: 391 updated += 1 392 else: 393 skipped += 1 394 code = quote.warnings[0] if quote.warnings else quote.quality_status 395 warnings.append(f"{row['ticker']}: {code}") 396 errors.append(quote.quality_status) 397 item_results.append({ 398 "instrument_id": row["instrument_id"], 399 "ticker": row["ticker"], 400 "status": "would_update" if req.dry_run and quote.quality_status == "fresh" else quote.quality_status, 401 "provider": quote.provider, 402 "provider_symbol": quote.provider_symbol, 403 "price_date": quote.fetched_at[:10] if quote.fetched_at else None, 404 "currency": quote.currency, 405 "attempts": attempt, 406 }) 407 coverage_total = len(all_rows) 408 valued = sum( 409 _has_fresh_price_for_target(conn, str(row["instrument_id"]), target_date) 410 or (req.dry_run and str(row["instrument_id"]) in successful_instruments) 411 for row in all_rows 412 ) 413 if valued == coverage_total and coverage_total > 0 and not req.dry_run: 414 try: 415 from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation 416 417 valuation = run_daily_market_valuation(conn, as_of=target) 418 if valuation.status != "complete": 419 warnings.append("portfolio_valuation_partial") 420 except RuntimeError as exc: 421 warnings.append(str(exc) if str(exc) == "market_job_already_running" else "portfolio_valuation_failed") 422 return MarketBatchUpdateResponse( 423 action="equity_update_quotes", 424 provider=req.provider, 425 mode="dry_run" if req.dry_run else "apply", 426 requested_at=requested_at, 427 completed_at=utc_now(), 428 total=len(row_states), 429 updated=updated, 430 skipped=skipped, 431 warnings=warnings, 432 errors=errors, 433 target_date=target, 434 result_price_date_from=min(result_dates) if result_dates else None, 435 result_price_date_to=max(result_dates) if result_dates else None, 436 eligible_total=coverage_total, 437 limit_applied=coverage_total > bounded_limit, 438 provider_calls=provider_calls, 439 would_update=would_update, 440 persistence_performed=not req.dry_run and updated > 0, 441 cached=cached, 442 processed=processed, 443 valued=valued, 444 coverage_total=coverage_total, 445 complete=coverage_total > 0 and valued == coverage_total, 446 results=item_results, 447 render_provider_calls=False, 448 ) 449 450 def get_equity_quote(conn: Connection, instrument_id: str) -> MarketQuoteResponse: 451 inst, mapping, warnings = _instrument_mapping(conn, instrument_id) 452 latest = conn.execute("SELECT * FROM market_prices WHERE instrument_id=? ORDER BY COALESCE(price_timestamp, created_at, price_date) DESC LIMIT 1", (instrument_id,)).fetchone() 453 provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None 454 if not latest: 455 return MarketQuoteResponse(currency=inst["currency"] if inst else None, provider_symbol=provider_symbol, quality_status="missing", warnings=warnings or ["Kurs fehlt"]) 456 return MarketQuoteResponse(latest_price=_decimal_text(latest["close"]), currency=latest["currency"], close=_decimal_text(latest["close"]), provider=latest["provider"], provider_symbol=provider_symbol, fetched_at=latest["price_timestamp"] or latest["created_at"], quality_status=latest["quality_status"] or "stale", warnings=warnings) 457 458 459 def get_equity_chart(conn: Connection, instrument_id: str, *, range: str = "1d", interval: str = "5m") -> MarketChartResponse: 460 inst, mapping, warnings = _instrument_mapping(conn, instrument_id) 461 points = get_equity_chart_points(conn, instrument_id, limit=390) 462 provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None 463 return _chart_response(points, currency=inst["currency"] if inst else "CHF", provider_symbol=provider_symbol, warnings=warnings) 464 465 466 def _normalized_candle_params(range_key: str, interval_key: str) -> tuple[str, str]: 467 allowed = { 468 "1d": {"5m"}, 469 "5d": {"15m"}, 470 "1mo": {"1d"}, 471 "6mo": {"1d"}, 472 "ytd": {"1d"}, 473 "1y": {"1d"}, 474 } 475 normalized_range = (range_key or "1d").lower() 476 if normalized_range not in allowed: 477 normalized_range = "1d" 478 normalized_interval = (interval_key or "").lower() 479 if normalized_interval not in allowed[normalized_range]: 480 normalized_interval = next(iter(allowed[normalized_range])) 481 return normalized_range, normalized_interval 482 483 484 def _candles_response_from_rows(rows: list[Any], *, instrument_id: str | None, symbol: str | None, range_key: str, interval_key: str, provider: str = "yfinance", quality_status: str = "fresh", warnings: list[str] | None = None) -> EquityCandlesResponse: 485 candles: list[Any] = [] 486 volumes: list[int | None] = [] 487 currency = None 488 exchange_timezone = None 489 fetched_at = None 490 for row in rows: 491 item = dict(row) if hasattr(row, "keys") else row 492 vol = item.get("volume") 493 volume_text = _decimal_text(vol) 494 candles.append({ 495 "time": str(item["timestamp"]), 496 "open": _decimal_text(item["open"]) or "", 497 "high": _decimal_text(item["high"]) or "", 498 "low": _decimal_text(item["low"]) or "", 499 "close": _decimal_text(item["close"]) or "", 500 "volume": volume_text, 501 }) 502 volumes.append(int(float(vol)) if vol not in (None, "") else None) 503 currency = currency or item.get("currency") 504 exchange_timezone = exchange_timezone or item.get("exchange_timezone") 505 fetched_at = item.get("fetched_at") or fetched_at 506 return EquityCandlesResponse(instrument_id=instrument_id, symbol=symbol, provider_symbol=symbol, range=range_key, interval=interval_key, provider=provider, quality_status=quality_status if candles else "missing", candles=candles, volume=volumes, currency=currency, exchange_timezone=exchange_timezone, fetched_at=fetched_at, warnings=warnings or ([] if candles else ["Zu wenig Kursdaten verfügbar."])) 507 508 509 def _yfinance_history(provider_symbol: str, *, range_key: str, interval_key: str) -> tuple[list[dict[str, Any]], str | None, str | None]: 510 try: 511 import yfinance as yf # type: ignore[import-not-found] 512 except Exception as exc: # pragma: no cover - environment dependent 513 raise RuntimeError("yfinance_not_installed") from exc 514 ticker = yf.Ticker(provider_symbol) 515 history = ticker.history(period=range_key, interval=interval_key, auto_adjust=False) 516 if history is None or getattr(history, "empty", True): 517 return [], None, None 518 info = getattr(ticker, "fast_info", {}) or {} 519 currency = None 520 try: 521 currency = info.get("currency") if hasattr(info, "get") else None 522 except Exception: 523 currency = None 524 exchange_timezone = str(getattr(history.index, "tz", "") or "") or None 525 candles: list[dict[str, Any]] = [] 526 for idx, row in history.iterrows(): 527 try: 528 open_v = Decimal(str(row["Open"])) 529 close_v = Decimal(str(row["Close"])) 530 low_v = Decimal(str(row["Low"])) 531 high_v = Decimal(str(row["High"])) 532 except (InvalidOperation, KeyError, ValueError): 533 continue 534 if any(v.is_nan() for v in (open_v, close_v, low_v, high_v)): 535 continue 536 timestamp = idx.isoformat() if hasattr(idx, "isoformat") else str(idx) 537 candles.append({"timestamp": timestamp, "open": open_v, "close": close_v, "low": low_v, "high": high_v, "volume": row.get("Volume"), "currency": currency, "exchange_timezone": exchange_timezone}) 538 return candles, currency, exchange_timezone 539 540 541 def get_equity_candles(conn: Connection, instrument_id: str, *, range: str = "1d", interval: str = "5m", refresh: bool = False) -> EquityCandlesResponse: 542 range_key, interval_key = _normalized_candle_params(range, interval) 543 inst, mapping, warnings = _instrument_mapping(conn, instrument_id) 544 provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None 545 if not provider_symbol: 546 return EquityCandlesResponse(instrument_id=instrument_id, symbol=None, provider_symbol=None, range=range_key, interval=interval_key, provider="yfinance", quality_status="missing_provider_symbol", currency=inst["currency"] if inst else None, warnings=warnings or ["Provider-Symbol fehlt"]) 547 if not refresh: 548 cached = get_equity_intraday_candles(conn, instrument_id, range_key=range_key, interval_key=interval_key, max_age_minutes=10) 549 if cached: 550 return _candles_response_from_rows(cached, instrument_id=instrument_id, symbol=str(provider_symbol), range_key=range_key, interval_key=interval_key, warnings=warnings) 551 try: 552 candles, currency, exchange_timezone = _yfinance_history(str(provider_symbol), range_key=range_key, interval_key=interval_key) 553 except Exception as exc: 554 return EquityCandlesResponse(instrument_id=instrument_id, symbol=str(provider_symbol), provider_symbol=str(provider_symbol), range=range_key, interval=interval_key, provider="yfinance", quality_status="provider_error", currency=inst["currency"] if inst else None, warnings=[str(exc)]) 555 if not candles: 556 return EquityCandlesResponse(instrument_id=instrument_id, symbol=str(provider_symbol), provider_symbol=str(provider_symbol), range=range_key, interval=interval_key, provider="yfinance", quality_status="missing", currency=currency or (inst["currency"] if inst else None), exchange_timezone=exchange_timezone, warnings=["Zu wenig Kursdaten verfügbar."]) 557 upsert_equity_intraday_candles(conn, instrument_id=instrument_id, provider="yfinance", provider_symbol=str(provider_symbol), range_key=range_key, interval_key=interval_key, candles=candles, currency=currency or (inst["currency"] if inst else None), exchange_timezone=exchange_timezone, quality_status="fresh") 558 conn.commit() 559 rows = get_equity_intraday_candles(conn, instrument_id, range_key=range_key, interval_key=interval_key, max_age_minutes=15) 560 return _candles_response_from_rows(rows, instrument_id=instrument_id, symbol=str(provider_symbol), range_key=range_key, interval_key=interval_key, warnings=warnings) 561 562 563 def refresh_equity_fx(conn: Connection, instrument_id: str) -> dict[str, object]: 564 inst = conn.execute("SELECT currency FROM instruments WHERE instrument_id=?", (instrument_id,)).fetchone() 565 if not inst: 566 raise HTTPException(status_code=404, detail="Instrument not found") 567 currencies: set[str] = set() 568 for value in [inst["currency"]]: 569 if value: 570 currencies.add(str(value).upper()) 571 latest_price = conn.execute("SELECT currency FROM market_prices WHERE instrument_id=? ORDER BY COALESCE(price_timestamp, created_at, price_date) DESC LIMIT 1", (instrument_id,)).fetchone() 572 if latest_price and latest_price["currency"]: 573 currencies.add(str(latest_price["currency"]).upper()) 574 tx_rows = conn.execute("SELECT transaction_id, currency_original, trade_date, fx_status, fx_rate_to_chf FROM transactions WHERE instrument_id=? AND COALESCE(is_voided,0)=0", (instrument_id,)).fetchall() 575 for row in tx_rows: 576 if row["currency_original"]: 577 currencies.add(str(row["currency_original"]).upper()) 578 updated = 0 579 warnings: list[str] = [] 580 from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider 581 from jarvis_finance.fx.rates import resolve_fx_rate_to_chf 582 for cur in sorted(currencies): 583 if cur == "CHF": 584 continue 585 latest_result = resolve_fx_rate_to_chf(conn, base_currency=cur, rate_date=None, providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], persist=True, resolve_fixed=True) 586 if latest_result.status == "ok": 587 updated += 1 588 elif latest_result.warning: 589 warnings.append(f"{cur}: {latest_result.warning}") 590 for row in tx_rows: 591 cur = str(row["currency_original"] or "").upper() 592 if cur == "CHF": 593 if row["fx_status"] != "not_needed" or not row["fx_rate_to_chf"]: 594 conn.execute("UPDATE transactions SET fx_rate_to_chf='1', fx_source='not_needed', fx_status='not_needed', updated_at=? WHERE transaction_id=?", (utc_now(), row["transaction_id"])) 595 updated += 1 596 continue 597 if cur and (row["fx_status"] == "missing" or not row["fx_rate_to_chf"]): 598 historical = resolve_fx_rate_to_chf(conn, base_currency=cur, rate_date=row["trade_date"], providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], persist=True, resolve_fixed=True) 599 if historical.status == "ok" and historical.rate is not None: 600 conn.execute("UPDATE transactions SET fx_rate_to_chf=?, fx_source=?, fx_status='ok', updated_at=? WHERE transaction_id=?", (format(historical.rate, "f"), historical.source, utc_now(), row["transaction_id"])) 601 updated += 1 602 elif historical.warning: 603 warnings.append(f"{cur} {row['trade_date']}: {historical.warning}") 604 conn.commit() 605 return {"action": "equity_fx_recheck", "instrument_id": instrument_id, "currencies": sorted(currencies), "updated": updated, "warnings": warnings[:10], "render_provider_calls": False} 606 607 608 def _crypto_asset(conn: Connection, asset_id: str): 609 asset = conn.execute("SELECT * FROM crypto_assets WHERE asset_id=?", (asset_id,)).fetchone() 610 if not asset: 611 raise HTTPException(status_code=404, detail="Crypto asset not found") 612 return asset 613 614 615 class BinanceClient: 616 base_url = "https://api.binance.com" 617 618 def __init__(self, opener=None, timeout_seconds: float = 8.0) -> None: 619 self.opener = opener or request.urlopen 620 self.timeout_seconds = timeout_seconds 621 622 def _json(self, path: str, params: dict[str, str]) -> object: 623 url = self.base_url + path + "?" + parse.urlencode(params) 624 try: 625 with self.opener(url, timeout=self.timeout_seconds) as response: # noqa: S310 - explicit user-triggered provider call 626 return json.loads(response.read().decode("utf-8")) 627 except error.HTTPError as exc: 628 if exc.code == 429: 629 raise RuntimeError("binance_rate_limited") from exc 630 if exc.code == 400: 631 raise RuntimeError("unsupported_pair") from exc 632 raise RuntimeError("binance_provider_error") from exc 633 except error.URLError as exc: 634 raise RuntimeError("binance_network_error") from exc 635 636 def ticker_24h(self, symbol: str) -> dict[str, object]: 637 data = self._json("/api/v3/ticker/24hr", {"symbol": symbol.upper()}) 638 return data if isinstance(data, dict) else {} 639 640 def klines(self, symbol: str, *, interval: str = "5m", limit: int = 288) -> list[list[object]]: 641 data = self._json("/api/v3/klines", {"symbol": symbol.upper(), "interval": interval, "limit": str(limit)}) 642 return data if isinstance(data, list) else [] 643 644 645 def refresh_crypto_quote(conn: Connection, asset_id: str, req: QuoteRefreshRequest) -> MarketQuoteResponse: 646 asset = _crypto_asset(conn, asset_id) 647 provider = (req.provider or "coingecko").lower() 648 if provider == "binance": 649 symbol = asset["binance_symbol"] if "binance_symbol" in asset.keys() else None 650 if not symbol: 651 return MarketQuoteResponse(currency="USD", provider="binance", quality_status="unsupported_pair", warnings=["Binance-Symbol fehlt"]) 652 try: 653 ticker = BinanceClient().ticker_24h(symbol) 654 price = Decimal(str(ticker.get("lastPrice"))) 655 ts = utc_now() 656 if not req.dry_run: 657 upsert_crypto_price_point(conn, asset_id=asset_id, timestamp=ts, price=price, currency="USD", provider="binance", provider_symbol=symbol, interval=req.interval, source_quality="fresh") 658 conn.commit() 659 return MarketQuoteResponse(latest_price=format(price, "f"), currency="USD", change_abs=_decimal_text(ticker.get("priceChange")), change_pct=_decimal_text(ticker.get("priceChangePercent")), high=_decimal_text(ticker.get("highPrice")), low=_decimal_text(ticker.get("lowPrice")), close=format(price, "f"), volume=_decimal_text(ticker.get("volume")), provider="binance", provider_symbol=symbol, fetched_at=ts, quality_status="fresh") 660 except Exception as exc: 661 return MarketQuoteResponse(currency="USD", provider="binance", provider_symbol=symbol, quality_status=_quality_from_error(str(exc), "provider_error"), warnings=[str(exc)]) 662 if not asset["coingecko_id"]: 663 return MarketQuoteResponse(currency=req.currency.upper(), provider="CoinGecko", quality_status="missing", warnings=["CoinGecko-ID fehlt"]) 664 quote: PriceQuote = CoinGeckoClient().get_crypto_price(asset["coingecko_id"], req.currency) 665 quality = quote.quality_status if quote.price is not None else _quality_from_error(quote.error_message, quote.quality_status) 666 ts = quote.provider_timestamp or utc_now() 667 if not req.dry_run: 668 store_crypto_price(conn, asset_id=asset_id, quote=quote) 669 if quote.price is not None: 670 upsert_crypto_price_point(conn, asset_id=asset_id, timestamp=ts, price=quote.price, currency=quote.currency, provider=quote.provider, provider_symbol=quote.coingecko_id, interval=req.interval, source_quality=quality) 671 conn.commit() 672 return MarketQuoteResponse(latest_price=_decimal_text(quote.price), currency=quote.currency, close=_decimal_text(quote.price), provider=quote.provider, provider_symbol=quote.coingecko_id, fetched_at=ts, quality_status=quality, warnings=[quote.error_message] if quote.error_message else []) 673 674 675 def refresh_crypto_quotes_batch(conn: Connection, req: QuoteRefreshRequest) -> MarketBatchUpdateResponse: 676 provider = (req.provider or "binance").lower() 677 if provider == "binance": 678 sql = "SELECT asset_id FROM crypto_assets WHERE is_active=1 AND binance_symbol IS NOT NULL AND binance_symbol!='' ORDER BY symbol LIMIT ?" 679 else: 680 sql = "SELECT asset_id FROM crypto_assets WHERE is_active=1 AND coingecko_id IS NOT NULL AND coingecko_id!='' ORDER BY symbol LIMIT ?" 681 rows = conn.execute(sql, (max(1, min(int(req.limit or 20), 100)),)).fetchall() 682 updated = skipped = 0 683 warnings: list[str] = [] 684 errors: list[str] = [] 685 for row in rows: 686 quote = refresh_crypto_quote(conn, row["asset_id"], req) 687 if quote.quality_status in {"fresh", "delayed", "stale"} and quote.latest_price is not None: 688 updated += 1 689 else: 690 skipped += 1 691 warnings.extend(quote.warnings) 692 if quote.quality_status in {"rate_limited", "provider_error"}: 693 errors.append(quote.quality_status) 694 return MarketBatchUpdateResponse(action="crypto_update_live_stats", provider=provider, total=len(rows), updated=updated, skipped=skipped, warnings=warnings[:10], errors=errors[:10], render_provider_calls=False) 695 696 697 def get_crypto_quote(conn: Connection, asset_id: str, *, currency: str = "CHF") -> MarketQuoteResponse: 698 asset = _crypto_asset(conn, asset_id) 699 latest = conn.execute("SELECT * FROM crypto_prices WHERE asset_id=? AND price_currency=? ORDER BY COALESCE(provider_timestamp, fetched_at) DESC LIMIT 1", (asset_id, currency.upper())).fetchone() 700 if not latest: 701 return MarketQuoteResponse(currency=currency.upper(), provider_symbol=asset["coingecko_id"], quality_status="missing", warnings=["Kurs fehlt"]) 702 return MarketQuoteResponse(latest_price=_decimal_text(latest["price"]), currency=latest["price_currency"], close=_decimal_text(latest["price"]), provider=latest["provider"], provider_symbol=latest["coingecko_id"], fetched_at=latest["provider_timestamp"] or latest["fetched_at"], quality_status=latest["quality_status"] or "stale") 703 704 705 def get_crypto_chart(conn: Connection, asset_id: str, *, range: str = "1d", interval: str = "5m", currency: str = "CHF") -> MarketChartResponse: 706 asset = _crypto_asset(conn, asset_id) 707 points = get_crypto_chart_points(conn, asset_id, currency=currency.upper(), limit=390) 708 return _chart_response(points, currency=currency.upper(), provider_symbol=asset["coingecko_id"]) __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23__HERMES_CWD_8d46a20096ed__