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