===== src/jarvis_finance/services/asset_price_refresh.py ===== 1 from __future__ import annotations 2 3 import hashlib 4 import json 5 import uuid 6 from datetime import UTC, datetime, timedelta 7 from pathlib import Path 8 from sqlite3 import Connection, SQLITE_DELETE, SQLITE_DENY, SQLITE_INSERT, SQLITE_OK, SQLITE_UPDATE 9 from typing import Any, Callable 10 11 from jarvis_finance.api.schemas.market import QuoteRefreshRequest 12 from jarvis_finance.audit.log import record_audit_event 13 from jarvis_finance.market.providers import CoinGeckoClient 14 from jarvis_finance.services.crypto_market_recovery import run_crypto_market_one_shot 15 from jarvis_finance.services.market_service import refresh_equity_quotes_batch 16 from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development 17 from jarvis_finance.storage.database import connect 18 19 SOURCES = ("equity", "crypto", "fx") 20 PROTECTED_TABLES = ( 21 "accounts", 22 "transactions", 23 "crypto_holdings", 24 "positions_snapshot", 25 "postfinance_snapshot_positions", 26 "truewealth_snapshot_positions", 27 ) 28 29 30 def _deny_protected_dml( 31 action: int, 32 table: str | None, 33 _column: str | None, 34 _database: str | None, 35 _trigger: str | None, 36 ) -> int: 37 if action in {SQLITE_INSERT, SQLITE_UPDATE, SQLITE_DELETE} and table in PROTECTED_TABLES: 38 return SQLITE_DENY 39 return SQLITE_OK 40 41 42 def _now() -> str: 43 return datetime.now(UTC).isoformat() 44 45 46 def _database_path(conn: Connection) -> str: 47 row = next((row for row in conn.execute("PRAGMA database_list") if str(row[1]) == "main"), None) 48 if not row or not str(row[2] or ""): 49 raise ValueError("asset_refresh_requires_persistent_database") 50 return str(Path(str(row[2])).resolve()) 51 52 53 def _protected_fingerprint(conn: Connection) -> str: 54 payload: dict[str, list[dict[str, Any]]] = {} 55 available = { 56 str(row[0]) 57 for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() 58 } 59 for table in PROTECTED_TABLES: 60 if table not in available: 61 continue 62 rows = conn.execute(f'SELECT * FROM "{table}" ORDER BY rowid').fetchall() 63 payload[table] = [dict(row) for row in rows] 64 return hashlib.sha256( 65 json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") 66 ).hexdigest() 67 68 69 def _status_payload(conn: Connection, job_id: str) -> dict[str, Any]: 70 job = conn.execute("SELECT * FROM asset_price_refresh_jobs WHERE job_id=?", (job_id,)).fetchone() 71 if not job: 72 raise ValueError("asset_price_refresh_job_not_found") 73 sources = [ 74 dict(row) 75 for row in conn.execute( 76 "SELECT * FROM asset_price_refresh_sources WHERE job_id=? ORDER BY CASE source WHEN 'equity' THEN 1 WHEN 'crypto' THEN 2 ELSE 3 END", 77 (job_id,), 78 ).fetchall() 79 ] 80 return { 81 "job_id": str(job["job_id"]), 82 "status": str(job["status"]), 83 "requested_at": str(job["requested_at"]), 84 "completed_at": str(job["completed_at"]) if job["completed_at"] else None, 85 "stale_before": str(job["stale_before"]), 86 "progress": {"completed": int(job["progress_completed"]), "total": int(job["progress_total"])}, 87 "sources": [ 88 { 89 "source": str(row["source"]), 90 "status": str(row["status"]), 91 "stale_candidates": int(row["stale_candidates"]), 92 "updated_count": int(row["updated_count"]), 93 "error_code": str(row["error_code"]) if row["error_code"] else None, 94 "started_at": str(row["started_at"]) if row["started_at"] else None, 95 "completed_at": str(row["completed_at"]) if row["completed_at"] else None, 96 } 97 for row in sources 98 ], 99 "wealth_snapshot_created": bool(job["wealth_snapshot_id"]), 100 "audit_recorded": bool(job["audit_id"]), 101 "provider_calls_on_read": False, 102 } 103 104 105 def create_asset_price_refresh_job(conn: Connection, *, stale_hours: int = 24) -> tuple[dict[str, Any], str]: 106 """Persist a queued job only. No provider call occurs before the HTTP response.""" 107 if conn.in_transaction: 108 raise ValueError("asset_refresh_requires_clean_transaction") 109 conn.execute("BEGIN IMMEDIATE") 110 try: 111 if conn.execute( 112 "SELECT 1 FROM asset_price_refresh_jobs WHERE status IN ('queued','running') LIMIT 1" 113 ).fetchone(): 114 raise ValueError("asset_price_refresh_job_already_running") 115 now = datetime.now(UTC) 116 job_id = f"asset-refresh-{uuid.uuid4().hex}" 117 stale_before = (now - timedelta(hours=max(1, min(stale_hours, 720)))).isoformat() 118 conn.execute( 119 """INSERT INTO asset_price_refresh_jobs( 120 job_id,status,requested_at,stale_before,progress_total,progress_completed 121 ) VALUES(?,'queued',?,?,3,0)""", 122 (job_id, now.isoformat(), stale_before), 123 ) 124 conn.executemany( 125 """INSERT INTO asset_price_refresh_sources( 126 job_id,source,status,stale_candidates,updated_count 127 ) VALUES(?,?,'pending',0,0)""", 128 [(job_id, source) for source in SOURCES], 129 ) 130 conn.commit() 131 except Exception: 132 if conn.in_transaction: 133 conn.rollback() 134 raise 135 return _status_payload(conn, job_id), _database_path(conn) 136 137 138 def _equity_source(conn: Connection, stale_before: str) -> tuple[int, int]: 139 response = refresh_equity_quotes_batch( 140 conn, 141 QuoteRefreshRequest( 142 provider="auto", 143 only_missing=True, 144 stale_before=stale_before, 145 limit=500, 146 max_retries=1, 147 pacing_seconds=0.15, 148 ), 149 ) 150 candidates = max(0, int(response.total) - int(response.cached)) 151 if response.errors and response.updated == 0 and candidates > 0: 152 raise RuntimeError("equity_provider_failed") 153 return candidates, int(response.updated) 154 155 156 def _crypto_source(conn: Connection, stale_before: str) -> tuple[int, int]: 157 candidates = int( 158 conn.execute( 159 """SELECT COUNT(*) FROM crypto_assets a 160 WHERE a.is_active=1 AND EXISTS( 161 SELECT 1 FROM crypto_holdings h WHERE h.asset_id=a.asset_id AND CAST(h.quantity AS REAL)<>0 162 ) AND NOT EXISTS( 163 SELECT 1 FROM crypto_prices p 164 WHERE p.asset_id=a.asset_id AND p.fetched_at>=? 165 AND p.quality_status='fresh' AND p.price IS NOT NULL 166 )""", 167 (stale_before,), 168 ).fetchone()[0] 169 ) 170 if candidates == 0: 171 return 0, 0 172 result = run_crypto_market_one_shot(conn, provider=CoinGeckoClient()) 173 if result.status != "complete": 174 raise RuntimeError("crypto_provider_" + result.status) 175 return candidates, int(result.price_stored) 176 177 178 def _fx_source(conn: Connection, stale_before: str) -> tuple[int, int]: 179 from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider 180 from jarvis_finance.fx.rates import resolve_fx_rate_to_chf 181 182 cutoff_date = stale_before[:10] 183 currencies = [ 184 str(row["currency"]).upper() 185 for row in conn.execute( 186 """SELECT DISTINCT upper(i.currency) currency 187 FROM instruments i 188 WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF' 189 AND NOT EXISTS( 190 SELECT 1 FROM fx_rates f 191 WHERE f.base_currency=upper(i.currency) AND f.quote_currency='CHF' 192 AND f.rate_date>=? AND f.quality_status IN ('fresh','ok') 193 ) 194 ORDER BY currency""", 195 (cutoff_date,), 196 ).fetchall() 197 ] 198 updated = 0 199 failures = 0 200 for currency in currencies: 201 try: 202 result = resolve_fx_rate_to_chf( 203 conn, 204 base_currency=currency, 205 rate_date=None, 206 providers=[FrankfurterFxProvider(), TwelveDataFxProvider()], 207 persist=True, 208 resolve_fixed=True, 209 ) 210 updated += int(result.status == "ok") 211 except Exception: 212 failures += 1 213 conn.commit() 214 if failures and updated == 0: 215 raise RuntimeError("fx_provider_failed") 216 return len(currencies), updated 217 218 219 DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], tuple[int, int]]] = { 220 "equity": _equity_source, 221 "crypto": _crypto_source, 222 "fx": _fx_source, 223 } 224 225 226 def run_asset_price_refresh( 227 db_path: str, 228 job_id: str, 229 *, 230 runners: dict[str, Callable[[Connection, str], tuple[int, int]]] | None = None, 231 ) -> None: 232 """Background worker with source isolation, stored progress and mutation guard.""" 233 conn = connect(db_path) 234 selected = runners or DEFAULT_RUNNERS 235 try: 236 conn.execute("BEGIN IMMEDIATE") 237 claimed = conn.execute( 238 """UPDATE asset_price_refresh_jobs 239 SET status='running' 240 WHERE job_id=? AND status='queued'""", 241 (job_id,), 242 ).rowcount 243 conn.commit() 244 if claimed != 1: 245 return 246 stale_before = str( 247 conn.execute( 248 "SELECT stale_before FROM asset_price_refresh_jobs WHERE job_id=?", 249 (job_id,), 250 ).fetchone()[0] 251 ) 252 protected_before = _protected_fingerprint(conn) 253 conn.set_authorizer(_deny_protected_dml) 254 completed = 0 255 failures = 0 256 for source in SOURCES: 257 started = _now() 258 conn.execute( 259 "UPDATE asset_price_refresh_sources SET status='running',started_at=? WHERE job_id=? AND source=?", 260 (started, job_id, source), 261 ) 262 conn.commit() 263 candidates = updated = 0 264 status = "complete" 265 error_code = None 266 try: 267 candidates, updated = selected[source](conn, stale_before) 268 if candidates == 0: 269 status = "skipped" 270 except Exception as exc: 271 if conn.in_transaction: 272 conn.rollback() 273 status = "failed" 274 failures += 1 275 error_code = str(exc)[:120] or type(exc).__name__ 276 completed += 1 277 conn.execute( 278 """UPDATE asset_price_refresh_sources 279 SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=? 280 WHERE job_id=? AND source=?""", 281 (status, candidates, updated, error_code, _now(), job_id, source), 282 ) 283 conn.execute( 284 "UPDATE asset_price_refresh_jobs SET progress_completed=? WHERE job_id=?", 285 (completed, job_id), 286 ) 287 conn.commit() 288 if _protected_fingerprint(conn) != protected_before: 289 raise RuntimeError("protected_holdings_or_transactions_mutated") 290 291 successful_sources = failures < len(SOURCES) 292 wealth_snapshot_id = None 293 if successful_sources: 294 model = build_modelled_wealth_development(conn, period="1m") 295 current = model.get("current") or {} 296 wealth_snapshot_id = f"wealth-refresh-{uuid.uuid4().hex}" 297 source_rows = [ 298 dict(row) 299 for row in conn.execute( 300 "SELECT source,status,stale_candidates,updated_count,error_code FROM asset_price_refresh_sources WHERE job_id=? ORDER BY source", 301 (job_id,), 302 ).fetchall() 303 ] 304 conn.execute( 305 """INSERT INTO aggregated_wealth_refresh_snapshots( 306 wealth_snapshot_id,job_id,captured_at,known_wealth_chf,quality_status,source_status_json 307 ) VALUES(?,?,?,?,?,?)""", 308 ( 309 wealth_snapshot_id, 310 job_id, 311 _now(), 312 current.get("value_chf"), 313 "complete" if failures == 0 else "partial", 314 json.dumps(source_rows, sort_keys=True), 315 ), 316 ) 317 final_status = "complete" if failures == 0 else "failed" if failures == len(SOURCES) else "partial" 318 audit_id = record_audit_event( 319 conn, 320 source="asset_price_refresh_job_v1", 321 action="asset_prices_refresh_completed", 322 entity_type="asset_price_refresh_job", 323 entity_id=job_id, 324 old_values={}, 325 new_values={ 326 "status": final_status, 327 "source_count": len(SOURCES), 328 "failed_source_count": failures, 329 "wealth_snapshot_created": bool(wealth_snapshot_id), 330 "holdings_mutated": False, 331 "transactions_mutated": False, 332 "trades_created": 0, 333 }, 334 created_by="system", 335 ) 336 conn.execute( 337 """UPDATE asset_price_refresh_jobs 338 SET status=?,completed_at=?,wealth_snapshot_id=?,audit_id=? 339 WHERE job_id=?""", 340 (final_status, _now(), wealth_snapshot_id, audit_id, job_id), 341 ) 342 conn.commit() 343 except Exception as exc: 344 if conn.in_transaction: 345 conn.rollback() 346 audit_id = record_audit_event( 347 conn, 348 source="asset_price_refresh_job_v1", 349 action="asset_prices_refresh_failed", 350 entity_type="asset_price_refresh_job", 351 entity_id=job_id, 352 old_values={}, 353 new_values={"status": "failed", "error_code": str(exc)[:120]}, 354 created_by="system", 355 ) 356 conn.execute( 357 "UPDATE asset_price_refresh_jobs SET status='failed',completed_at=?,audit_id=? WHERE job_id=?", 358 (_now(), audit_id, job_id), 359 ) 360 conn.commit() 361 finally: 362 conn.close() 363 364 365 def asset_price_refresh_status(conn: Connection, job_id: str) -> dict[str, Any]: 366 """Stored status only: no provider call, write or lazy refresh.""" 367 return _status_payload(conn, job_id) ===== src/jarvis_finance/api/routers/market.py ===== 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) ===== src/jarvis_finance/api/schemas/market.py ===== 1 from __future__ import annotations 2 3 from pydantic import BaseModel, ConfigDict, Field 4 from typing import Literal 5 6 7 class AssetPriceRefreshRequest(BaseModel): 8 model_config = ConfigDict(extra="forbid") 9 stale_hours: int = Field(default=24, ge=1, le=720) 10 11 12 class AssetPriceRefreshSourceStatus(BaseModel): 13 model_config = ConfigDict(extra="forbid") 14 source: Literal["equity", "crypto", "fx"] 15 status: Literal["pending", "running", "complete", "failed", "skipped"] 16 stale_candidates: int 17 updated_count: int 18 error_code: str | None 19 started_at: str | None 20 completed_at: str | None 21 22 23 class AssetPriceRefreshJobResponse(BaseModel): 24 model_config = ConfigDict(extra="forbid") 25 job_id: str 26 status: Literal["queued", "running", "complete", "partial", "failed"] 27 requested_at: str 28 completed_at: str | None 29 stale_before: str 30 progress: dict[str, int] 31 sources: list[AssetPriceRefreshSourceStatus] 32 wealth_snapshot_created: bool 33 audit_recorded: bool 34 provider_calls_on_read: Literal[False] 35 36 37 class MarketStatusResponse(BaseModel): 38 equity_latest_update: str | None = None 39 crypto_latest_update: str | None = None 40 equity_cached_points: int = 0 41 crypto_cached_points: int = 0 42 mapped_equity_instruments: int = 0 43 mapped_crypto_assets: int = 0 44 render_provider_calls: bool = False 45 warnings: list[str] = Field(default_factory=list) 46 47 48 class QuoteRefreshRequest(BaseModel): 49 provider: str = "auto" 50 currency: str = "CHF" 51 range: str = "1d" 52 interval: str = "5m" 53 limit: int = Field(default=100, ge=1, le=500) 54 price_date: str | None = None 55 only_missing: bool = True 56 stale_before: str | None = None 57 max_retries: int = Field(default=2, ge=0, le=3) 58 pacing_seconds: float = Field(default=0.6, ge=0, le=5) 59 max_parallelism: int = Field(default=3, ge=1, le=4) 60 dry_run: bool = False 61 62 63 class MarketBatchUpdateResponse(BaseModel): 64 action: str 65 provider: str 66 mode: str = "apply" 67 requested_at: str | None = None 68 completed_at: str | None = None 69 total: int = 0 70 updated: int = 0 71 skipped: int = 0 72 warnings: list[str] = Field(default_factory=list) 73 errors: list[str] = Field(default_factory=list) 74 target_date: str | None = None 75 result_price_date_from: str | None = None 76 result_price_date_to: str | None = None 77 eligible_total: int = 0 78 limit_applied: bool = False 79 provider_calls: int = 0 80 would_update: int = 0 81 persistence_performed: bool = False 82 cached: int = 0 83 processed: int = 0 84 valued: int = 0 85 coverage_total: int = 0 86 complete: bool = False 87 results: list[dict[str, str | int | bool | None]] = Field(default_factory=list) 88 render_provider_calls: bool = False 89 90 91 class MarketQuoteResponse(BaseModel): 92 latest_price: str | None = None 93 currency: str | None = None 94 change_abs: str | None = None 95 change_pct: str | None = None 96 open: str | None = None 97 high: str | None = None 98 low: str | None = None 99 close: str | None = None 100 volume: str | None = None 101 provider: str | None = None 102 provider_symbol: str | None = None 103 fetched_at: str | None = None 104 quality_status: str = "missing" 105 warnings: list[str] = Field(default_factory=list) 106 chart_points: list[dict[str, str | None]] = Field(default_factory=list) 107 108 109 class ChartPoint(BaseModel): 110 timestamp: str 111 price: str 112 currency: str 113 provider: str | None = None 114 quality_status: str | None = None 115 116 117 class MarketChartResponse(BaseModel): 118 latest_price: str | None = None 119 currency: str | None = None 120 change_abs: str | None = None 121 change_pct: str | None = None 122 open: str | None = None 123 high: str | None = None 124 low: str | None = None 125 close: str | None = None 126 volume: str | None = None 127 provider: str | None = None 128 provider_symbol: str | None = None 129 fetched_at: str | None = None 130 quality_status: str = "missing" 131 chart_points: list[ChartPoint] = Field(default_factory=list) 132 warnings: list[str] = Field(default_factory=list) 133 134 135 class EquityCandle(BaseModel): 136 time: str 137 open: str 138 high: str 139 low: str 140 close: str 141 volume: str | None = None 142 143 144 class EquityCandlesResponse(BaseModel): 145 instrument_id: str | None = None 146 symbol: str | None = None 147 provider_symbol: str | None = None 148 range: str = "1d" 149 interval: str = "5m" 150 provider: str = "yfinance" 151 quality_status: str = "missing" 152 candles: list[EquityCandle] = Field(default_factory=list) 153 volume: list[int | None] = Field(default_factory=list) 154 currency: str | None = None 155 exchange_timezone: str | None = None 156 fetched_at: str | None = None 157 warnings: list[str] = Field(default_factory=list) ===== src/jarvis_finance/services/market_service.py ===== 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__