caudal/programa/herramientas/serie_btc_eur.py

"""
serie_btc_eur.py — Genera cartera/serie_btc_eur.csv: el cierre diario del
bitcoin en euros desde 2011, que la app lleva empaquetado.

    python3 herramientas/serie_btc_eur.py

Por qué empaquetado: las fuentes gratuitas «al vuelo» solo dan el último año
(CoinGecko) o los últimos 720 días (Kraken), y el historial de un monedero
puede empezar mucho antes. Además, así la app no le cuenta a nadie qué días
te interesan.

Fuentes:
  • Desde el 25-abr-2016: cierre diario BTC/EUR de Bitstamp (vela UTC).
  • Antes: cierre BTC/USD de Bitstamp × cambio USD→EUR del BCE del mismo día
    (o del último día hábil anterior: el BCE no publica en fin de semana).
Contraste (se imprime; si se desvía demasiado, el script para):
  • Los últimos 720 días contra el cierre BTC/EUR de Kraken.
  • El tramo calculado en dólares contra el BTC/EUR real de Bitstamp del
    primer año en que coinciden, para medir el error de ese método.

Este precio es solo una referencia (valorar lo minado o recibido, estimar un
importe que falta, avisar de primas raras). El coste fiscal de una compra es
siempre lo que pagaste de verdad.
"""

from __future__ import annotations

import csv
import datetime as dt
import json
import statistics
import sys
import time
import urllib.request
from pathlib import Path

DESTINO = Path(__file__).resolve().parent.parent / "cartera" / "serie_btc_eur.csv"
INICIO_USD = dt.date(2011, 9, 13)
INICIO_EUR = dt.date(2016, 4, 25)
DESVIO_MAXIMO = 0.03          # 3 % de mediana entre fuentes: más, algo va mal


def _json(url: str):
    req = urllib.request.Request(url, headers={"User-Agent": "cartera-serie/1"})
    for intento in range(4):
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                return json.load(r)
        except OSError:
            if intento == 3:
                raise
            time.sleep(3)


def _dia(ts) -> dt.date:
    return dt.datetime.fromtimestamp(int(ts), dt.timezone.utc).date()


def bitstamp(par: str, desde: dt.date, hasta: dt.date) -> dict[dt.date, float]:
    cierres: dict[dt.date, float] = {}
    inicio = int(dt.datetime(desde.year, desde.month, desde.day, tzinfo=dt.timezone.utc).timestamp())
    while True:
        velas = _json(f"https://www.bitstamp.net/api/v2/ohlc/{par}/?step=86400&limit=1000&start={inicio}")
        velas = velas["data"]["ohlc"]
        nuevas = 0
        for v in velas:
            d = _dia(v["timestamp"])
            if desde <= d <= hasta and float(v["close"]) > 0 and d not in cierres:
                cierres[d] = float(v["close"])
                nuevas += 1
        if not velas or not nuevas or _dia(velas[-1]["timestamp"]) >= hasta:
            return cierres
        inicio = int(velas[-1]["timestamp"]) + 86400
        time.sleep(0.5)


def bce_usd_eur(desde: dt.date, hasta: dt.date) -> dict[dt.date, float]:
    r = _json(f"https://api.frankfurter.dev/v1/{desde.isoformat()}..{hasta.isoformat()}?from=USD&to=EUR")
    return {dt.date.fromisoformat(d): v["EUR"] for d, v in r["rates"].items()}


def kraken_eur() -> dict[dt.date, float]:
    r = _json("https://api.kraken.com/0/public/OHLC?pair=XBTEUR&interval=1440")
    if r.get("error"):
        raise RuntimeError(f"Kraken: {r['error']}")
    velas = next(iter(r["result"].values()))
    return {_dia(v[0]): float(v[4]) for v in velas}


def _en_dolares(usd: dict, cambio: dict, dia: dt.date) -> float | None:
    if dia not in usd:
        return None
    for atras in range(8):                      # fin de semana y festivos del BCE
        c = cambio.get(dia - dt.timedelta(days=atras))
        if c:
            return usd[dia] * c
    return None


def _desvio(a: dict, b: dict) -> tuple[float, float, int]:
    comunes = sorted(set(a) & set(b))
    difs = [abs(a[d] / b[d] - 1) for d in comunes if b[d]]
    if not difs:
        return 0.0, 0.0, 0
    return statistics.median(difs), max(difs), len(difs)


def main() -> int:
    ayer = dt.datetime.now(dt.timezone.utc).date() - dt.timedelta(days=1)
    print("Bitstamp BTC/EUR…")
    eur = bitstamp("btceur", INICIO_EUR, ayer)
    print("Bitstamp BTC/USD…")
    usd = bitstamp("btcusd", INICIO_USD, INICIO_EUR + dt.timedelta(days=365))
    print("BCE USD→EUR…")
    cambio = bce_usd_eur(INICIO_USD - dt.timedelta(days=10), INICIO_EUR + dt.timedelta(days=365))
    print("Kraken BTC/EUR (contraste)…")
    kraken = kraken_eur()

    serie: dict[dt.date, tuple[float, str]] = {}
    dia = INICIO_USD
    while dia < INICIO_EUR:
        v = _en_dolares(usd, cambio, dia)
        if v:
            serie[dia] = (v, "bitstamp-usd*bce")
        dia += dt.timedelta(days=1)
    for d, v in eur.items():
        serie[d] = (v, "bitstamp-eur")

    # Contrastes
    solapado = {d: _en_dolares(usd, cambio, d) for d in usd if d >= INICIO_EUR}
    solapado = {d: v for d, v in solapado.items() if v}
    med_usd, max_usd, n_usd = _desvio(solapado, eur)
    med_kr, max_kr, n_kr = _desvio({d: v for d, (v, _) in serie.items()}, kraken)
    print(f"Método dólares×BCE contra BTC/EUR real: mediana {med_usd:.2%}, máximo {max_usd:.2%} ({n_usd} días)")
    print(f"Bitstamp contra Kraken:                  mediana {med_kr:.2%}, máximo {max_kr:.2%} ({n_kr} días)")
    if med_usd > DESVIO_MAXIMO or med_kr > DESVIO_MAXIMO or n_kr < 300:
        print("✗ Las fuentes no cuadran entre sí: no escribo nada.")
        return 1

    huecos = []
    dia = min(serie)
    while dia <= max(serie):
        if dia not in serie:
            huecos.append(dia)
        dia += dt.timedelta(days=1)
    tmp = DESTINO.with_suffix(".tmp")
    with tmp.open("w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow(["fecha", "cierre_eur", "fuente"])
        for d in sorted(serie):
            v, fuente = serie[d]
            w.writerow([d.isoformat(), f"{v:.2f}", fuente])
    tmp.replace(DESTINO)
    print(f"✓ {len(serie)} días, del {min(serie)} al {max(serie)} · {len(huecos)} días sin dato "
          f"(la app usa el anterior) → {DESTINO}")
    return 0


if __name__ == "__main__":
    sys.exit(main())