430 ORDER BY price_date DESC, created_at DESC LIMIT 1 431 """, 432 (instrument_id, provider, price_date), 433 ).fetchone() 434 435 436 def _corporate_action_status(conn: Connection, *, instrument_id: str, price_date: str, provider: str, close: Decimal | None) -> str: 437 if close is None or close <= 0: 438 return "not_checked" 439 prev = _previous_price(conn, instrument_id=instrument_id, price_date=price_date, provider=provider) 440 if not prev or not prev["close"]: 441 return "not_checked" 442 old = Decimal(str(prev["close"])) 443 if old <= 0: 444 return "not_checked" 445 change = abs((close - old) / old) 446 if change > Decimal("0.25"): 447 create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="corporate_action_suspected", message="Large local price move detected; corporate action review required before high-confidence valuation.", evidence={"previous_price_date": prev["price_date"], "price_date": price_date}, fingerprint="corporate_action_suspected") 448 create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id="split_or_corporate_action_review_required", message="Split/corporate-action review required; no automatic split correction is applied.", evidence={"previous_price_date": prev["price_date"], "price_date": price_date}, fingerprint="split_or_corporate_action_review_required") 449 update_instrument_metadata(conn, instrument_id=instrument_id, corporate_action_status="suspected", split_or_corporate_action_review_required=True, note="Automatic C2 heuristic detected >25% local price move; review required.") 450 return "suspected" 451 return "none_known" 452 453 454 def _economic_price_payload( 455 *, 456 instrument_id: str, 457 provider: str, 458 provider_symbol: str | None, 459 provider_market: str | None, 460 price_type: str, 461 close: Decimal | None, 462 adjusted_close: Decimal | None, 463 currency: str, 464 provider_timestamp: str, 465 quality_status: str, 466 source_reference: str, 467 ) -> dict[str, str | None]: 468 return { 469 "provider": provider.lower(), 470 "instrument_id": instrument_id, 471 "provider_symbol": provider_symbol, 472 "provider_market": provider_market, 473 "price_type": price_type, 474 "close": format(close, "f") if close is not None else "", 475 "adjusted_close": format(adjusted_close, "f") if adjusted_close is not None else None, 476 "currency": currency.upper(), 477 "provider_timestamp": provider_timestamp, 478 "source_reference": source_reference, 479 "quality_status": quality_status, 480 } 481 482 483 def _store_economic_price_observation( 484 conn: Connection, 485 *, 486 instrument_id: str, 487 provider: str, 488 provider_symbol: str | None, 489 provider_market: str | None, 490 price_type: str, 491 close: Decimal | None, 492 adjusted_close: Decimal | None, 493 currency: str, 494 provider_timestamp: str, 495 quality_status: str, 496 source_reference: str | None, 497 job_reference: str | None, 498 created_at: str, 499 ) -> str: 500 origin_reference = source_reference or ":".join( 501 part for part in (provider.lower(), provider_symbol or "", provider_market or "") if part 502 ) 503 payload = _economic_price_payload( 504 instrument_id=instrument_id, 505 provider=provider, 506 provider_symbol=provider_symbol, 507 provider_market=provider_market, 508 price_type=price_type, 509 close=close, 510 adjusted_close=adjusted_close, 511 currency=currency, 512 provider_timestamp=provider_timestamp, 513 quality_status=quality_status, 514 source_reference=origin_reference, 515 ) 516 payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":")) 517 payload_hash = hashlib.sha256(payload_json.encode()).hexdigest() 518 source_observation_id = stable_id( 519 "market-source-observation", 520 provider.lower(), 521 instrument_id, 522 provider_symbol or "", 523 price_type, 524 provider_timestamp, 525 ) 526 same = conn.execute( 527 """SELECT observation_id FROM market_price_observations 528 WHERE source_observation_id=? AND economic_payload_hash=?""", 529 (source_observation_id, payload_hash), 530 ).fetchone() 531 if same: 532 return str(same["observation_id"]) 533 predecessor = conn.execute( 534 """SELECT observation_id,payload_version,economic_payload_json 535 FROM market_price_observations WHERE source_observation_id=? 536 ORDER BY payload_version DESC LIMIT 1""", 537 (source_observation_id,), 538 ).fetchone() 539 version = int(predecessor["payload_version"]) + 1 if predecessor else 1 540 observation_id = stable_id("market-observation", source_observation_id, payload_hash) 541 conn.execute( 542 """INSERT INTO market_price_observations( 543 observation_id,source_observation_id,payload_version,supersedes_observation_id, 544 instrument_id,provider,provider_symbol,provider_market,price_type,close,adjusted_close, 545 currency,provider_timestamp,source_reference,quality_status,economic_payload_json, 546 economic_payload_hash,created_at,job_reference 547 ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", 548 ( 549 observation_id, source_observation_id, version, 550 str(predecessor["observation_id"]) if predecessor else None, 551 instrument_id, provider, provider_symbol, provider_market, price_type, 552 payload["close"], payload["adjusted_close"], payload["currency"], provider_timestamp, 553 origin_reference, quality_status, payload_json, payload_hash, created_at, job_reference, 554 ), 555 ) 556 if predecessor: 557 record_audit_event( 558 conn, 559 source="market_price_observation_v2", 560 action="market_price_observation_corrected", 561 entity_type="market_price_observation", 562 entity_id=observation_id, 563 old_values={"supersedes_observation_id": predecessor["observation_id"]}, 564 new_values={"source_observation_id": source_observation_id, "payload_version": version}, 565 confirmed=True, 566 created_by="system", 567 ) 568 return observation_id 569 570 571 def store_market_price( 572 conn: Connection, 573 *, 574 instrument_id: str, 575 price_date: str, 576 close: Decimal | None, 577 currency: str, 578 provider: str, 579 provider_symbol: str | None, 580 provider_market: str | None = None, 581 price_timestamp: str | None = None, 582 adjusted_close: Decimal | None = None, 583 quality_status: str = "fresh", 584 error_message: str | None = None, 585 fetched_at: str | None = None, 586 price_type: str = "unadjusted_close", 587 run_id: str | None = None, 588 ) -> str: 589 now = utc_now() 590 fetched = fetched_at or now 591 existing = conn.execute( 592 "SELECT market_price_id, close, currency, quality_status FROM market_prices WHERE instrument_id=? AND price_date=? AND provider=?", 593 (instrument_id, price_date, provider), 594 ).fetchone() 595 corp_status = _corporate_action_status(conn, instrument_id=instrument_id, price_date=price_date, provider=provider, close=close) if quality_status == "fresh" else "not_checked" 596 provider_observed_at = price_timestamp or price_date 597 _store_economic_price_observation( 598 conn, 599 instrument_id=instrument_id, 600 provider=provider, 601 provider_symbol=provider_symbol, 602 provider_market=provider_market, 603 price_type=price_type, 604 close=close, 605 adjusted_close=adjusted_close, 606 currency=currency, 607 provider_timestamp=provider_observed_at, 608 quality_status=quality_status, 609 source_reference=None, 610 job_reference=run_id, 611 created_at=now, 612 ) 613 market_price_id = str(existing["market_price_id"]) if existing else stable_id( 614 "marketprice", instrument_id, price_date, provider, provider_symbol or "", now 615 ) 616 conn.execute( 617 """ 618 INSERT INTO market_prices( 619 market_price_id, instrument_id, price_date, price_timestamp, close, adjusted_close, 620 currency, provider, provider_symbol, quality_status, created_at, provider_market, error_message, 621 corporate_action_status, fetched_at, price_type, run_id 622 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 623 ON CONFLICT(instrument_id, price_date, provider) DO UPDATE SET 624 price_timestamp=excluded.price_timestamp, 625 close=excluded.close, 626 adjusted_close=excluded.adjusted_close, 627 currency=excluded.currency, 628 provider_symbol=excluded.provider_symbol, 629 quality_status=excluded.quality_status, 630 created_at=excluded.created_at, 631 provider_market=excluded.provider_market, 632 error_message=excluded.error_message, 633 corporate_action_status=excluded.corporate_action_status, 634 fetched_at=excluded.fetched_at, 635 price_type=excluded.price_type, 636 run_id=excluded.run_id 637 """, 638 (market_price_id, instrument_id, price_date, price_timestamp, format(close, "f") if close is not None else "", format(adjusted_close, "f") if adjusted_close is not None else None, currency.upper(), provider, provider_symbol, quality_status, now, provider_market, error_message, corp_status, fetched, price_type, run_id), 639 ) 640 if existing and ( 641 str(existing["close"] or "") != (format(close, "f") if close is not None else "") 642 or str(existing["currency"] or "") != currency.upper() 643 ): 644 record_audit_event( 645 conn, 646 source="daily_market_fx_v1", 647 action="market_price_provider_correction", 648 entity_type="market_price", 649 entity_id=str(existing["market_price_id"]), 650 old_values={"close": existing["close"], "currency": existing["currency"], "quality_status": existing["quality_status"]}, 651 new_values={"close": format(close, "f") if close is not None else None, "currency": currency.upper(), "quality_status": quality_status, "run_id": run_id}, 652 confirmed=True, 653 created_by="system", 654 ) 655 if quality_status in {"missing", "stale", "error", "conflict"} or close is None: 656 rule = "stale_market_price" if quality_status == "stale" else "missing_market_price" 657 create_alert(conn, priority="warnung", category="market_data", entity_type="instrument", entity_id=instrument_id, rule_id=rule, message="Instrument local market price is not fresh.", evidence={"provider": provider, "provider_symbol": provider_symbol, "price_date": price_date, "quality_status": quality_status}, fingerprint=f"{rule}:{provider}:{provider_symbol}") 658 elif quality_status == "fresh": 659 _resolve_market_alerts(conn, instrument_id=instrument_id) 660 conn.commit() 661 return market_price_id 662 663 664 def refresh_market_prices( 665 conn: Connection, 666 *, 667 provider: EquityPriceProvider, 668 asset_class: str, 669 price_date: str | None = None, 670 only_missing: bool = False, 671 only_stale: bool = False, 672 only_isin: str | None = None, 673 limit: int | None = None, 674 dry_run: bool = False, 675 ) -> MarketPriceRefreshResult: 676 asset = asset_class.lower() 677 result = MarketPriceRefreshResult(asset_class=asset, dry_run=dry_run) 678 query = """ 679 SELECT m.*, i.asset_class, i.isin, i.instrument_status AS current_instrument_status, i.valuation_policy AS current_valuation_policy FROM instrument_price_mappings m 680 JOIN instruments i ON i.instrument_id=m.instrument_id 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( 210 quote, 211 currency=actual_currency or expected_currency, 212 price_timestamp=quote.price_timestamp or quote_date.isoformat(), 213 quality_status="fresh", 214 ) 215 216 217 def refresh_equity_quote(conn: Connection, instrument_id: str, req: QuoteRefreshRequest) -> MarketQuoteResponse: 218 inst, mapping, warnings = _instrument_mapping(conn, instrument_id) 219 if warnings: 220 return MarketQuoteResponse(provider_symbol=None, currency=inst["currency"] if inst else None, quality_status="missing", warnings=warnings) 221 mapping_data = dict(mapping) if mapping else { 222 "provider": inst["data_provider_primary"] or "auto", 223 "provider_symbol": inst["provider_symbol"], 224 "provider_market": inst["exchange"], 225 "trading_currency": inst["currency"], 226 "currency": inst["currency"], 227 } 228 provider_symbol = str(mapping_data["provider_symbol"]) 229 provider_name = (req.provider or "auto").lower() 230 target = _effective_market_date(req.price_date) 231 quote = equity_price_provider_by_name(provider_name).get_price(provider_symbol, price_date=target.isoformat()) 232 quote = _validate_historical_quote(quote, mapping=mapping_data, target=target, requested_provider=provider_name) 233 quality = quote.quality_status if quote.close is not None else _quality_from_error(quote.error_message, quote.quality_status) 234 ts = quote.price_timestamp or target.isoformat() 235 economic_observation_created = False 236 if not req.dry_run and quote.close is not None and quality == "fresh": 237 observation_count_before = conn.execute( 238 "SELECT COUNT(*) FROM market_price_observations WHERE instrument_id=?", 239 (instrument_id,), 240 ).fetchone()[0] 241 store_market_price( 242 conn, 243 instrument_id=instrument_id, 244 price_date=ts[:10], 245 close=quote.close, 246 currency=quote.currency or inst["currency"] or "CHF", 247 provider=quote.provider, 248 provider_symbol=quote.provider_symbol or provider_symbol, 249 provider_market=quote.provider_market or mapping_data["provider_market"], 250 price_timestamp=ts, 251 adjusted_close=quote.adjusted_close, 252 quality_status=quality, 253 error_message=None, 254 ) 255 economic_observation_created = conn.execute( 256 "SELECT COUNT(*) FROM market_price_observations WHERE instrument_id=?", 257 (instrument_id,), 258 ).fetchone()[0] > observation_count_before 259 upsert_equity_price_point( 260 conn, 261 instrument_id=instrument_id, 262 timestamp=ts, 263 price=quote.close, 264 currency=quote.currency or inst["currency"] or "CHF", 265 provider=quote.provider, 266 provider_symbol=quote.provider_symbol or provider_symbol, 267 interval=req.interval, 268 source_quality=quality, 269 ) 270 conn.commit() 271 return MarketQuoteResponse( 272 latest_price=_decimal_text(quote.close), 273 currency=quote.currency or inst["currency"], 274 close=_decimal_text(quote.close), 275 provider=quote.provider, 276 provider_symbol=quote.provider_symbol or provider_symbol, 277 fetched_at=ts, 278 quality_status=quality, 279 economic_observation_created=economic_observation_created, 280 warnings=warnings + ([quote.error_message] if quote.error_message else []), 281 ) 282 283 284 def _persistent_database_path(conn: Connection) -> str | None: 285 row = next((row for row in conn.execute("PRAGMA database_list") if str(row[1]) == "main"), None) 286 return str(row[2]) if row and str(row[2] or "") else None 287 288 289 def _parallel_equity_worker( 290 db_path: str, instrument_id: str, req: QuoteRefreshRequest 291 ) -> tuple[MarketQuoteResponse, int]: 292 worker = connect(db_path) 293 worker.execute("PRAGMA busy_timeout=10000") 294 try: 295 attempt = 0 296 quote: MarketQuoteResponse | None = None 297 while attempt <= req.max_retries: 298 attempt += 1 299 quote = refresh_equity_quote(worker, instrument_id, req) 300 if quote.latest_price is not None and quote.quality_status == "fresh": 301 break 302 if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries: 303 break 304 time.sleep(min(2 ** (attempt - 1), 4)) 305 assert quote is not None 306 return quote, attempt 307 finally: 308 worker.close() 309 310 311 def refresh_equity_quotes_batch(conn: Connection, req: QuoteRefreshRequest) -> MarketBatchUpdateResponse: 312 requested_at = utc_now() 313 target_date = _effective_market_date(req.price_date) 314 target = target_date.isoformat() 315 all_rows = conn.execute( 316 """ 317 SELECT DISTINCT i.instrument_id,i.name,i.ticker 318 FROM instruments i 319 JOIN instrument_price_mappings m 320 ON m.instrument_id=i.instrument_id AND m.mapping_status='mapped' 321 WHERE i.asset_class IN ('stock','equity','etf') 322 AND i.is_active=1 323 AND COALESCE(i.instrument_status,'active') NOT IN ('inactive','delisted','suspended','merged') 324 AND COALESCE(i.valuation_policy,'')!='exclude_from_auto_price_update' 325 AND m.provider_symbol IS NOT NULL AND m.provider_symbol!='' 326 ORDER BY i.name 327 """, 328 ).fetchall() 329 row_states = [ 330 ( 331 row, 332 _has_fresh_price_for_target( 333 conn, 334 str(row["instrument_id"]), 335 target_date, 336 stale_before=req.stale_before, 337 ), 338 ) 339 for row in all_rows 340 ] 341 row_states.sort(key=lambda item: (item[1], str(item[0]["name"] or ""))) 342 bounded_limit = max(1, min(int(req.limit or 100), 500)) 343 row_states = row_states[:bounded_limit] 344 updated = economic_updated = skipped = cached = processed = 0 345 would_update = provider_calls = 0 346 successful_instruments: set[str] = set() 347 result_dates: list[str] = [] 348 warnings: list[str] = [] 349 errors: list[str] = [] 350 item_results: list[dict[str, str | int | bool | None]] = [] 351 last_call_at = 0.0 352 parallel_results: dict[str, tuple[MarketQuoteResponse, int]] = {} 353 db_path = _persistent_database_path(conn) 354 uncached_rows = [row for row, has_fresh in row_states if not (req.only_missing and has_fresh)] 355 if db_path and not req.dry_run and req.max_parallelism > 1 and len(uncached_rows) > 1: 356 with ThreadPoolExecutor(max_workers=min(req.max_parallelism, len(uncached_rows))) as executor: 357 futures = { 358 executor.submit(_parallel_equity_worker, db_path, str(row["instrument_id"]), req): str(row["instrument_id"]) 359 for row in uncached_rows 360 } 361 for future in as_completed(futures): 362 instrument_id = futures[future] 363 try: 364 parallel_results[instrument_id] = future.result() 365 except Exception as exc: 366 parallel_results[instrument_id] = ( 367 MarketQuoteResponse(quality_status="provider_error", warnings=[type(exc).__name__]), 368 1, 369 ) 370 for row, has_fresh in row_states: 371 if req.only_missing and has_fresh: 372 cached += 1 373 skipped += 1 374 item_results.append({"instrument_id": row["instrument_id"], "ticker": row["ticker"], "status": "cached", "attempts": 0}) 375 continue 376 parallel = parallel_results.get(str(row["instrument_id"])) 377 if parallel: 378 quote, attempt = parallel 379 processed += attempt 380 provider_calls += attempt 381 else: 382 elapsed = time.monotonic() - last_call_at 383 if last_call_at and elapsed < req.pacing_seconds: 384 time.sleep(req.pacing_seconds - elapsed) 385 attempt = 0 386 quote: MarketQuoteResponse | None = None 387 while attempt <= req.max_retries: 388 attempt += 1 389 processed += 1 390 provider_calls += 1 391 last_call_at = time.monotonic() 392 try: 393 quote = refresh_equity_quote(conn, row["instrument_id"], req) 394 except Exception as exc: 395 quote = MarketQuoteResponse( 396 quality_status="provider_error", 397 warnings=[type(exc).__name__], 398 ) 399 if quote.latest_price is not None and quote.quality_status == "fresh": 400 break 401 if quote.quality_status not in {"rate_limited", "network_error"} or attempt > req.max_retries: 402 break 403 time.sleep(min(2 ** (attempt - 1), 4)) 404 assert quote is not None 405 if quote.latest_price is not None and quote.quality_status == "fresh": 406 successful_instruments.add(str(row["instrument_id"])) 407 if quote.fetched_at: 408 result_dates.append(quote.fetched_at[:10]) 409 if req.dry_run: 410 would_update += 1 411 else: 412 updated += 1 413 if quote.economic_observation_created: 414 economic_updated += 1 415 else: 416 skipped += 1 417 code = quote.warnings[0] if quote.warnings else quote.quality_status 418 warnings.append(f"{row['ticker']}: {code}") 419 errors.append(quote.quality_status) 420 item_results.append({ 421 "instrument_id": row["instrument_id"], 422 "ticker": row["ticker"], 423 "status": "would_update" if req.dry_run and quote.quality_status == "fresh" else quote.quality_status, 424 "provider": quote.provider, 425 "provider_symbol": quote.provider_symbol, 426 "price_date": quote.fetched_at[:10] if quote.fetched_at else None, 427 "currency": quote.currency, 428 "attempts": attempt, 429 }) 430 coverage_total = len(all_rows) 431 valued = sum( 432 _has_fresh_price_for_target(conn, str(row["instrument_id"]), target_date) 433 or (req.dry_run and str(row["instrument_id"]) in successful_instruments) 434 for row in all_rows 435 ) 436 if valued == coverage_total and coverage_total > 0 and not req.dry_run: 437 try: 438 from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation 439 440 valuation = run_daily_market_valuation(conn, as_of=target) 441 if valuation.status != "complete": 442 warnings.append("portfolio_valuation_partial") 443 except RuntimeError as exc: 444 warnings.append(str(exc) if str(exc) == "market_job_already_running" else "portfolio_valuation_failed") 445 return MarketBatchUpdateResponse( 446 action="equity_update_quotes", 447 provider=req.provider, 448 mode="dry_run" if req.dry_run else "apply", 449 requested_at=requested_at, 450 completed_at=utc_now(), 451 total=len(row_states), 452 updated=updated, 453 economic_updated=economic_updated, 454 skipped=skipped, 455 warnings=warnings, 456 errors=errors, 457 target_date=target, 458 result_price_date_from=min(result_dates) if result_dates else None, 459 result_price_date_to=max(result_dates) if result_dates else None, 460 eligible_total=coverage_total, 461 limit_applied=coverage_total > bounded_limit, 462 provider_calls=provider_calls, 463 would_update=would_update, 464 persistence_performed=not req.dry_run and updated > 0, 465 cached=cached, 466 processed=processed, 467 valued=valued, 468 coverage_total=coverage_total, 469 complete=coverage_total > 0 and valued == coverage_total, 470 results=item_results, 471 render_provider_calls=False, 472 ) 473 474 def get_equity_quote(conn: Connection, instrument_id: str) -> MarketQuoteResponse: 475 inst, mapping, warnings = _instrument_mapping(conn, instrument_id) 476 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() 477 provider_symbol = (mapping["provider_symbol"] if mapping else inst["provider_symbol"]) if inst else None 478 if not latest: 479 return MarketQuoteResponse(currency=inst["currency"] if inst else None, provider_symbol=provider_symbol, quality_status="missing", warnings=warnings or ["Kurs fehlt"]) 480 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) __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23.1__HERMES_CWD_8d46a20096ed__