150 lines
5.1 KiB
Python
150 lines
5.1 KiB
Python
"""Einheitliche Preis-API — eine Quelle der Wahrheit für Spot + Händler."""
|
|
import requests
|
|
import sqlite3
|
|
from typing import Optional
|
|
from .db import get_db
|
|
|
|
OZ = 31.1035 # Gramm pro Unze
|
|
|
|
|
|
def get_exchange_rate() -> float:
|
|
"""EUR/USD Kurs aus DB, Fallback 0.92."""
|
|
try:
|
|
conn = get_db()
|
|
row = conn.execute(
|
|
"SELECT rate FROM exchange_rates ORDER BY timestamp DESC LIMIT 1"
|
|
).fetchone()
|
|
conn.close()
|
|
return row["rate"] if row else 0.92
|
|
except Exception:
|
|
return 0.92
|
|
|
|
|
|
def fetch_spot() -> Optional[dict]:
|
|
"""
|
|
Holt Gold/Silber Spotpreise.
|
|
Gibt EUR und USD zurück — immer beide Einheiten klar benannt.
|
|
"""
|
|
try:
|
|
r_gold = requests.get("https://api.gold-api.com/price/XAU", timeout=10)
|
|
r_silver = requests.get("https://api.gold-api.com/price/XAG", timeout=10)
|
|
|
|
gold_usd = r_gold.json().get("price", 0)
|
|
silver_usd = r_silver.json().get("price", 0)
|
|
eur_usd = get_exchange_rate()
|
|
|
|
gold_eur = gold_usd * eur_usd
|
|
silver_eur = silver_usd * eur_usd
|
|
|
|
return {
|
|
# USD pro Unze
|
|
"gold_usd_oz": gold_usd,
|
|
"silver_usd_oz": silver_usd,
|
|
# EUR pro Unze
|
|
"gold_eur_oz": gold_eur,
|
|
"silver_eur_oz": silver_eur,
|
|
# EUR pro Gramm (für Validierung + Portfolio)
|
|
"gold_eur_g": gold_eur / OZ,
|
|
"silver_eur_g": silver_eur / OZ,
|
|
# Meta
|
|
"eur_usd": eur_usd,
|
|
"ratio": gold_usd / silver_usd if silver_usd > 0 else 0,
|
|
}
|
|
except Exception as e:
|
|
return None
|
|
|
|
|
|
def save_spot(spot: dict):
|
|
"""Spotpreis in DB speichern."""
|
|
conn = get_db()
|
|
conn.execute("""
|
|
INSERT INTO spot_prices (gold_usd, silver_usd, gold_eur, silver_eur, eur_usd)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
""", (
|
|
spot["gold_usd_oz"], spot["silver_usd_oz"],
|
|
spot["gold_eur_oz"], spot["silver_eur_oz"],
|
|
spot["eur_usd"],
|
|
))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def get_latest_spot() -> Optional[dict]:
|
|
"""Letzten gespeicherten Spotpreis aus DB."""
|
|
conn = get_db()
|
|
row = conn.execute("""
|
|
SELECT gold_usd, silver_usd, gold_eur, silver_eur, eur_usd, timestamp
|
|
FROM spot_prices ORDER BY timestamp DESC LIMIT 1
|
|
""").fetchone()
|
|
conn.close()
|
|
if not row:
|
|
return None
|
|
eur_usd = row["eur_usd"] or 0.92
|
|
gold_eur_oz = row["gold_eur"] or (row["gold_usd"] * eur_usd)
|
|
silver_eur_oz = row["silver_eur"] or (row["silver_usd"] * eur_usd)
|
|
return {
|
|
"gold_usd_oz": row["gold_usd"],
|
|
"silver_usd_oz": row["silver_usd"],
|
|
"gold_eur_oz": gold_eur_oz,
|
|
"silver_eur_oz": silver_eur_oz,
|
|
"gold_eur_g": gold_eur_oz / OZ,
|
|
"silver_eur_g": silver_eur_oz / OZ,
|
|
"eur_usd": eur_usd,
|
|
"ratio": row["gold_usd"] / row["silver_usd"] if row["silver_usd"] else 0,
|
|
"timestamp": row["timestamp"],
|
|
}
|
|
|
|
|
|
def get_best_dealer_prices(hours: int = 6) -> dict:
|
|
"""
|
|
Bester Ankaufpreis für 1oz Münzen der letzten N Stunden.
|
|
Einfach, direkt, keine komplexe Filterlogik.
|
|
"""
|
|
conn = get_db()
|
|
|
|
# Gold: Krügerrand bevorzugen, dann andere Standardmünzen (Median statt Max)
|
|
gold_rows = conn.execute("""
|
|
SELECT product, buy_price, sell_price, weight_g,
|
|
COALESCE(buy_price, sell_price) / weight_g AS ppg
|
|
FROM gold_prices
|
|
WHERE weight_g BETWEEN 30 AND 32
|
|
AND COALESCE(buy_price, sell_price) > 0
|
|
AND timestamp >= datetime('now', ?)
|
|
ORDER BY ppg
|
|
""", (f"-{hours} hours",)).fetchall()
|
|
|
|
# Krügerrand bevorzugen, sonst Median
|
|
gold = next(
|
|
(r for r in gold_rows if "krügerrand" in r["product"].lower() or "kruger" in r["product"].lower()),
|
|
gold_rows[len(gold_rows)//2] if gold_rows else None
|
|
)
|
|
|
|
# Silber: Standardmünzen (Maple, Philharmoniker etc.), Median
|
|
silver_rows = conn.execute("""
|
|
SELECT product, buy_price, sell_price, weight_g,
|
|
COALESCE(buy_price, sell_price) / weight_g AS ppg
|
|
FROM silver_prices
|
|
WHERE weight_g BETWEEN 30 AND 32
|
|
AND COALESCE(buy_price, sell_price) > 0
|
|
AND timestamp >= datetime('now', ?)
|
|
ORDER BY ppg
|
|
""", (f"-{hours} hours",)).fetchall()
|
|
|
|
silver = next(
|
|
(r for r in silver_rows if any(k in r["product"].lower()
|
|
for k in ["maple", "philharmoniker", "känguru", "britannia", "kruger"])),
|
|
silver_rows[len(silver_rows)//2] if silver_rows else None
|
|
)
|
|
|
|
conn.close()
|
|
def _price(row):
|
|
if not row: return 0
|
|
return row["buy_price"] if row["buy_price"] else (row["sell_price"] if row["sell_price"] else round(row["ppg"] * 31.1035, 2))
|
|
return {
|
|
"gold_ppg": gold["ppg"] if gold else 0,
|
|
"gold_oz": _price(gold),
|
|
"gold_product": gold["product"] if gold else "—",
|
|
"silver_ppg": silver["ppg"] if silver else 0,
|
|
"silver_oz": _price(silver),
|
|
"silver_product": silver["product"] if silver else "—",
|
|
}
|