#!/usr/bin/env python3
"""下载Binance 2017-2026 BTC日线数据并保存"""
import requests, zipfile, io, pandas as pd
from concurrent.futures import ThreadPoolExecutor

years_months = []
for y in range(2017, 2027):
    for m in range(1, 13):
        if y == 2017 and m < 8: continue
        if y == 2026 and m > 7: continue
        years_months.append(f"{y}-{m:02d}")

def dl(ym):
    try:
        r = requests.get(f"https://data.binance.vision/data/spot/monthly/klines/BTCUSDT/1d/BTCUSDT-1d-{ym}.zip", timeout=30)
        if r.status_code != 200: return None
        z = zipfile.ZipFile(io.BytesIO(r.content))
        with z.open(z.namelist()[0]) as f:
            return pd.read_csv(f, header=None)
    except: return None

with ThreadPoolExecutor(max_workers=5) as ex:
    results = list(ex.map(dl, years_months))

dfs = []
for ym, df in zip(years_months, results):
    if df is not None:
        df = df.iloc[:, :6].copy()
        df.columns = ["ts","o","h","l","c","v"]
        df[["o","h","l","c","v"]] = df[["o","h","l","c","v"]].astype(float)
        dfs.append(df)

all_data = pd.concat(dfs).sort_values("ts").drop_duplicates(subset="ts")
all_data["dt"] = pd.to_datetime(all_data["ts"], unit="ms", errors="coerce")
all_data = all_data.dropna(subset=["dt"])
all_data = all_data[(all_data["dt"] >= "2010-01-01") & (all_data["dt"] <= "2030-01-01")]
all_data = all_data.set_index("dt").drop(columns=["ts"])

print(f"日线: {len(all_data)}根  {all_data.index[0].date()} ~ {all_data.index[-1].date()}")
print(f"价格: ${all_data['l'].min():.0f} ~ ${all_data['h'].max():.0f}")

all_data.to_parquet("/root/quant_pipeline/data/btc_daily.parquet")

weekly = all_data.resample("W").agg({"o":"first","h":"max","l":"min","c":"last","v":"sum"}).dropna()
monthly = all_data.resample("M").agg({"o":"first","h":"max","l":"min","c":"last","v":"sum"}).dropna()
weekly.to_parquet("/root/quant_pipeline/data/btc_weekly.parquet")
monthly.to_parquet("/root/quant_pipeline/data/btc_monthly.parquet")

print(f"周线: {len(weekly)}根 月线: {len(monthly)}根")
print("✅ 全部保存成功!")