pipeline/live_data.py

"""
live_data.py — 全特征 API 实时更新（不依赖本地 CSV）
=====================================================
数据源:
  1. FRED API       — 美国宏观、价格、利率、ISM 子指数 (~90)
  2. Yahoo Finance  — 商品价格、汇率、波动率指数 (~25)
  3. World Bank API — 全球大宗商品价格 (~40)
  4. akshare        — 中国宏观 M1/M2/PMI (~9)
  5. CFTC           — 持仓报告（自动下载 CSV）(~30)
  6. EIA API        — 美国能源库存/产量 (~8)
  7. GPR Index      — 地缘政治风险指数 (Caldara & Iacoviello)
  8. 派生特征         — 从原始特征自动计算 (~55)

用法:
  python live_data.py               # 全量更新
  python live_data.py --test        # 测试各 API 连通性
  python live_data.py --source fred # 只更新某个数据源
"""

import pandas as pd
import numpy as np
import os, json, warnings, argparse, time, requests
from datetime import datetime, timedelta

warnings.filterwarnings('ignore')
os.chdir(r'e:\大三下\比赛\花旗杯\模型')

FRED_KEY = 'fc02a6e6a359a4cc16f0f1752d258011'
EIA_KEY  = '9Nv5PhLREMmmKeo0zJ2U3Zu21Bntf8DfhEKBpi55'
OUTPUT_DIR = 'output'


# ═══════════════════════════════════════════════════════════
# 1. FRED API — 美国宏观 + 价格 + 利率 + ISM
# ═══════════════════════════════════════════════════════════

FRED_MAP = {
    # ── 价格 ──
    'WTI_spot':             'DCOILWTICO',
    'Brent_spot':           'DCOILBRENTEU',
    'natgas_spot_henry':    'DHHNGSP',
    'gold_spot':            'GOLDAMGBD228NLBM',

    # ── 需求/宏观 ──
    'pmi_us_mfg':           'MANEMP',
    'ipi_us':               'INDPRO',
    'nonfarm_us':           'PAYEMS',
    'usd_index':            'DTWEXBGS',
    'cpi_us':               'CPIAUCSL',
    'fed_funds_rate':       'FEDFUNDS',
    'yield_spread_10y2y':   'T10Y2Y',
    'us_10y_yield':         'DGS10',

    # ── 风险 ──
    'vix':                  'VIXCLS',

    # ── 供给 ──
    'us_oil_inventory_total': 'WCESTUS1',
    'us_crude_production':    'MCRFPUS2',

    # ── 利率/收益率 ──
    'libor_usd_3m':         'USD3MTD156N',
    '美国:国债收益率:3年':      'DGS3',
    '美国:国债收益率:7年':      'DGS7',
    '美国:国债收益率:10年':     'DGS10',
    '美国:国债收益率:30年':     'DGS30',
    '美国:国债到期收益率(以通胀为标的):7年': 'DFII7',
    '美国:国债收益率利差:10年-2年': 'T10Y2Y',

    # ── 货币 ──
    '美国:M1:非季调':         'M1NS',
    'us_m2_yoy':            'WM2NS',
    '美国:货币乘数:M1:非季调':  'MULT',
    '美国:货币乘数:M2:非季调':  'M2REAL',

    # ── ISM 制造业 PMI 子指数（40个）──
    '美国:ISM:制造业PMI':              'NAPM',
    '美国:ISM:制造业PMI:新订单':         'NAPMNOI',
    '美国:ISM:制造业PMI:自有库存':        'NAPMII',
    '美国:ISM:制造业PMI:客户库存':        'NAPMCI',
    '美国:ISM:制造业PMI:就业':           'NAPMEI',
    '美国:ISM:制造业PMI:物价':           'NAPMPRI',
    '美国:ISM:制造业PMI:产出':           'NAPMPI',
    '美国:ISM:制造业PMI:供应商交付':       'NAPMSDI',
    '美国:ISM:制造业PMI:新出口订单':       'NAPMEX',
    '美国:ISM:制造业PMI:进口':           'NAPMIMP',
    '美国:ISM:服务业PMI':              'NMFBAI',
    '美国:ISM:服务业PMI:就业':           'NMFEI',
    '美国:ISM:服务业PMI:新订单':          'NMFNOI',
    '美国:ISM:服务业PMI:物价':           'NMFPI',
    '美国:ISM:服务业PMI:商业活动':         'NMFBAI',
    '美国:ISM:服务业PMI:供应商交付':        'NMFSDI',
    '美国:ISM:服务业PMI:库存':           'NMFII',
    '美国:ISM:服务业PMI:订单库存':         'NMFOBI',
    '美国:ISM:服务业PMI:新出口订单':        'NMFEXI',
    '美国:ISM:服务业PMI:进口':           'NMFIMI',
    'pmi_us_svc':                     'NMFBAI',
    'employment_us_季调':              'PAYEMS',

    # ── GDP ──
    'gdp_us_yoy':            'A191RL1Q225SBEA',
    'gdp_japan_yoy':         'JPNRGDPEXP',
    'gdp_germany_yoy':       'CLVMNACSCAB1GQDE',
    'gdp_uk_yoy':            'CLVMNACSCAB1GQUK',
    'gdp_australia_yoy':     'NAEXKP01AUQ189S',

    # ── PPI/CPI ──
    'ppi_us_yoy':            'PPIACO',
    'ppi_us_同比':            'PPIACO',
    'cpi_japan':             'JPNCPIALLMINMEI',
    'ppi_japan':             'JPNPROINDMISMEI',
    'ppi_eurozone':          'EA19PRMNIG01GYSAM',
    'ipi_europe':            'EA19PRINTO01GYSAM',
    'ppi_russia':            'RUSCPIALLMINMEI',

    # ── 欧洲/日本宏观 ──
    'pmi_eu_mfg':            'BSCICP03EZM460S',
    'ez_econ_sentiment':     'CSCICP03EZM460S',
    'ez_consumer_conf':      'CSCICP03EZM460S',
    'pmi_japan':             'BSCICP03JPM460S',
    'pmi_canada':            'BSCICP03CAM460S',

    # ── OFR ──
    '美国:OFR金融压力指数':    'STLFSI2',

    # ── 美联储资产负债表 (FRED有汇总) ──
    'fed_balance_sheet_total': 'WALCL',

    # ── 国际利率 (LIBOR→SOFR替代) ──
    'libor_3m_英镑LIBOR3M':   'GBP3MTD156N',
    'libor_3m_日元LIBOR3M':   'JPY3MTD156N',
    'eurlibor_3m':           'EUR3MTD156N',

    # ── 美元指数(新兴市场) ──
    'usd_index_em':          'DTWEXEMEGS',

    # ── 美国石油消费/需求 (FRED镜像EIA) ──
    'us_oil_consumption':    'MTTIMUS2',
    'demand_us':             'MTTIMUS2',

    # ── 美国贸易 ──
    'us_crude_export_yoy':   'MCREXUS2',
    'us_crude_import_yoy':   'MCRIMUS2',

    # ── 美联储资产负债表明细 (已验证可用) ──
    'fed_balance_sheet_存款机构储备(不包括FHLB存款)':  'WRESBAL',
    'fed_balance_sheet_证券回购协议(包括官方储备中的其他债权)': 'WLRRAL',
    'fed_balance_sheet_对联邦政府的可支票存款':          'WDFOL',
    'fed_balance_sheet_联邦储备银行股票':             'TREAST',
    'fed_balance_sheet_联邦住房贷款银行(FHLB)借款':    'WSHOMCB',

    # ── 储备/需求 ──
    'reserve_us':            'TOTRESNS',

    # ── HIBOR ──
    'hibor_3m':              'HKONTD156N',

    # ── 中国PMI (FRED OECD proxy) ──
    'pmi_china':             'CHNPMINDMISMEI',
}


def fetch_fred_bulk(start='2000-01-01'):
    """批量从 FRED 拉取所有映射的指标。"""
    from fredapi import Fred
    fred = Fred(api_key=FRED_KEY)
    end = datetime.now().strftime('%Y-%m-%d')

    results = {}
    total = len(FRED_MAP)
    print(f"\n[FRED] 拉取 {total} 个指标...")

    for i, (name, sid) in enumerate(FRED_MAP.items()):
        try:
            data = fred.get_series(sid, observation_start=start, observation_end=end)
            if data is not None and len(data) > 0:
                monthly = data.resample('ME').last().dropna()
                results[name] = monthly
                print(f"  [{i+1}/{total}] ✓ {name:<35} {len(monthly)} 月")
            else:
                print(f"  [{i+1}/{total}] ✗ {name:<35} 无数据")
        except Exception as e:
            print(f"  [{i+1}/{total}] ✗ {name:<35} {str(e)[:40]}")

    df = pd.DataFrame(results) if results else pd.DataFrame()
    print(f"  FRED 完成: {len(results)}/{total}")
    return df


# ═══════════════════════════════════════════════════════════
# 2. YAHOO FINANCE — 商品/汇率/波动率
# ═══════════════════════════════════════════════════════════

YFINANCE_MAP = {
    # === 汇率 ===
    'usd_cny':              'CNY=X',
    'usd_jpy':              'JPY=X',

    # === 波动率 ===
    'vix_nasdaq':           '^VXN',
    'ovx_crude_vol':        '^OVX',
    '纳斯达克100波动率指数':    '^VXN',

    # === 贵金属 ===
    'iron_ore_spot':        'TIOC.SI',
    'nickel_price':         '^SPGSNKP',
    '现货价(伦敦市场):黄金:美元': 'GC=F',
    '现货价(伦敦市场):白银:美元': 'SI=F',
    'COMEX微型黄金':          'MGC=F',

    # === 大宗商品（替代 World Bank）===
    '全球:实际市场价格:铜':     'HG=F',
    '全球:实际市场价格:铝':     'ALI=F',
    '全球:实际市场价格:锌':     'ZINC.L',
    '全球:实际市场价格:玉米':   'ZC=F',
    '全球:实际市场价格:大米':   'ZR=F',
    '全球:实际市场价格:棉花':   'CT=F',
    '全球:实际市场价格:豆粕':   'ZM=F',
    '全球:实际市场价格:豆油':   'ZL=F',
    '全球:实际市场价格:糖(CESE第11号合同最近的未来头寸)': 'SB=F',
    '全球:实际市场价格:天然气(印尼出口日本)': 'TTF=F',
    '全球:商品价格:原油:三大市场平均值': 'CL=F',

    # === 农产品 ===
    'agri_prices_大豆':     'ZS=F',
    'agri_prices_小麦':     'ZW=F',
    'agri_prices_燕麦':     'ZO=F',

    # === 更多商品（替代 World Bank 剩余）===
    'natgas_index':         'NG=F',        # 天然气期货
    '全球:实际市场价格:牛肉':  'LE=F',        # 活牛期货
    '全球:实际市场价格:橄榄油': 'ZL=F',        # 豆油近似
    '全球:实际市场价格:鱼粉':  'ZM=F',        # 豆粕近似

    # === 汇率proxy（用于供给/能源衍生）===
    'micex_brent':          'USDRUB=X',    # USD/RUB作为MICEX-Brent proxy
}


def fetch_yfinance(start='2000-01-01'):
    """从 Yahoo Finance 拉取价格/汇率/商品数据（替代 World Bank）。"""
    import yfinance as yf

    tickers = {k: v for k, v in YFINANCE_MAP.items() if v is not None}
    total = len(tickers)
    print(f"\n[YFinance] 拉取 {total} 个指标（含商品替代 World Bank）...")

    results = {}
    for i, (name, ticker) in enumerate(tickers.items()):
        try:
            data = yf.download(ticker, start=start, progress=False, auto_adjust=True)
            if data is not None and len(data) > 0:
                close = data['Close']
                if hasattr(close, 'columns'):
                    close = close.iloc[:, 0]
                monthly = close.resample('ME').last().dropna()
                results[name] = monthly
                print(f"  [{i+1}/{total}] ✓ {name:<35} {len(monthly)} 月")
            else:
                print(f"  [{i+1}/{total}] ✗ {name:<35} 无数据")
        except Exception as e:
            print(f"  [{i+1}/{total}] ✗ {name:<35} {str(e)[:40]}")

    df = pd.DataFrame(results) if results else pd.DataFrame()
    print(f"  YFinance 完成: {len(results)}/{total}")
    return df


# ═══════════════════════════════════════════════════════════
# 3. (已合并到 YFinance) World Bank 商品 → yfinance 期货替代
# ═══════════════════════════════════════════════════════════

def fetch_worldbank():
    """World Bank 已被 yfinance 商品期货替代，返回空 DataFrame。"""
    print(f"\n[World Bank] 已合并到 YFinance 商品期货，跳过")
    return pd.DataFrame()


# ═══════════════════════════════════════════════════════════
# 4. AKSHARE — 中国宏观
# ═══════════════════════════════════════════════════════════

def fetch_akshare():
    """从 akshare 拉取中国宏观数据（M1/M2/PMI/GDP/CPI/PPI）。"""
    print(f"\n[akshare] 拉取中国宏观数据...")

    results = {}
    try:
        import akshare as ak

        # 中国 M1/M2 (货币供应量)
        try:
            ms = ak.macro_china_money_supply()
            if ms is not None and len(ms) > 0:
                ms['date'] = pd.to_datetime(ms.iloc[:, 0], errors='coerce')
                ms = ms.dropna(subset=['date']).set_index('date').sort_index()
                for col in ms.columns:
                    ccl = col.lower()
                    if 'M2' in col and '同比' in col:
                        results['中国:M2:同比'] = pd.to_numeric(ms[col], errors='coerce')
                    elif 'M1' in col and '同比' in col:
                        results['中国:M1:同比'] = pd.to_numeric(ms[col], errors='coerce')
                    elif 'M2' in col and '数量' in col:
                        results['中国:M2'] = pd.to_numeric(ms[col], errors='coerce')
                    elif 'M1' in col and '数量' in col:
                        results['中国:M1'] = pd.to_numeric(ms[col], errors='coerce')
                for k in ['中国:M1', '中国:M2', '中国:M1:同比', '中国:M2:同比']:
                    if k in results:
                        print(f"  ✓ {k:<25} {results[k].notna().sum()} 月")
        except Exception as e:
            print(f"  ✗ money_supply: {str(e)[:40]}")

        # 中国 PMI
        try:
            pmi = ak.macro_china_pmi()
            if pmi is not None and len(pmi) > 0:
                pmi['date'] = pd.to_datetime(pmi.iloc[:, 0], errors='coerce')
                pmi = pmi.dropna(subset=['date']).set_index('date').sort_index()
                for col in pmi.columns:
                    if '制造业' in col and '指数' in col:
                        results['pmi_china_mfg'] = pd.to_numeric(pmi[col], errors='coerce')
                        print(f"  ✓ pmi_china_mfg           {results['pmi_china_mfg'].notna().sum()} 月")
                    elif '非制造业' in col and '指数' in col:
                        results['pmi_china_nonsvc'] = pd.to_numeric(pmi[col], errors='coerce')
                        print(f"  ✓ pmi_china_nonsvc        {results['pmi_china_nonsvc'].notna().sum()} 月")
        except Exception as e:
            print(f"  ✗ pmi_china: {str(e)[:40]}")

        # 中国 CPI (月度)
        try:
            cpi = ak.macro_china_cpi_monthly()
            if cpi is not None and len(cpi) > 0:
                cpi['date'] = pd.to_datetime(cpi.iloc[:, 0], errors='coerce')
                cpi = cpi.dropna(subset=['date']).set_index('date').sort_index()
                if len(cpi.columns) > 1:
                    results['cpi_china'] = pd.to_numeric(cpi.iloc[:, 0], errors='coerce')
                    print(f"  ✓ cpi_china               {results['cpi_china'].notna().sum()} 月")
        except Exception as e:
            print(f"  ✗ cpi_china: {str(e)[:40]}")

        # 中国 PPI
        try:
            ppi = ak.macro_china_ppi()
            if ppi is not None and len(ppi) > 0:
                ppi['date'] = pd.to_datetime(ppi.iloc[:, 0], errors='coerce')
                ppi = ppi.dropna(subset=['date']).set_index('date').sort_index()
                if len(ppi.columns) > 1:
                    results['ppi_china_yoy'] = pd.to_numeric(ppi.iloc[:, 1], errors='coerce')
                    print(f"  ✓ ppi_china_yoy           {results['ppi_china_yoy'].notna().sum()} 月")
        except Exception as e:
            print(f"  ✗ ppi_china: {str(e)[:40]}")

        # 中国 GDP
        try:
            gdp = ak.macro_china_gdp()
            if gdp is not None and len(gdp) > 0:
                gdp['date'] = pd.to_datetime(gdp.iloc[:, 0], errors='coerce')
                gdp = gdp.dropna(subset=['date']).set_index('date').sort_index()
                for col in gdp.columns:
                    if '同比增长' in col:
                        results['gdp_china_yoy'] = pd.to_numeric(gdp[col], errors='coerce')
                        results['gdp_china_growth_yoy'] = results['gdp_china_yoy']
                        print(f"  ✓ gdp_china_yoy           {results['gdp_china_yoy'].notna().sum()} 季")
                        break
        except Exception as e:
            print(f"  ✗ gdp_china: {str(e)[:40]}")

        # SHIBOR 3M
        try:
            shibor = ak.rate_interbank(market="上海银行同业拆借市场",
                                       symbol="Shibor人民币", indicator="3月")
            if shibor is not None and len(shibor) > 0:
                shibor['date'] = pd.to_datetime(shibor.iloc[:, 0], errors='coerce')
                shibor = shibor.set_index('date').sort_index()
                series = pd.to_numeric(shibor.iloc[:, 0], errors='coerce')
                monthly = series.resample('ME').last().dropna()
                results['shibor_3m'] = monthly
                print(f"  ✓ shibor_3m               {len(monthly)} 月")
        except Exception as e:
            print(f"  ✗ shibor_3m: {str(e)[:40]}")

    except ImportError:
        print("  ✗ akshare 未安装 (pip install akshare)")
    except Exception as e:
        print(f"  ✗ akshare 错误: {e}")

    df = pd.DataFrame(results) if results else pd.DataFrame()
    print(f"  akshare 完成: {len(results)} 指标")
    return df


# ═══════════════════════════════════════════════════════════
# 5. CFTC 持仓报告（自动下载）
# ═══════════════════════════════════════════════════════════

def fetch_cftc():
    """从 CFTC 官网自动下载持仓报告。"""
    print(f"\n[CFTC] 下载持仓报告...")

    results = {}
    try:
        # CFTC Disaggregated Futures-Only reports
        year = datetime.now().year
        url = f"https://www.cftc.gov/dea/newcot/{year}fut.zip"

        import zipfile, io
        resp = requests.get(url, timeout=30)
        if resp.status_code == 200:
            z = zipfile.ZipFile(io.BytesIO(resp.content))
            fname = z.namelist()[0]
            df_cftc = pd.read_csv(z.open(fname))

            # Filter for crude oil
            oil = df_cftc[df_cftc['Market_and_Exchange_Names'].str.contains(
                'CRUDE OIL', case=False, na=False)]

            if len(oil) > 0:
                oil['date'] = pd.to_datetime(oil['As_of_Date_In_Form_YYMMDD'],
                                             format='%y%m%d', errors='coerce')
                oil = oil.set_index('date').sort_index()

                # Extract key positioning columns
                col_map = {
                    'ice_wti_mm_long': 'M_Money_Positions_Long_All',
                    'ice_wti_mm_short': 'M_Money_Positions_Short_All',
                }
                for name, col in col_map.items():
                    if col in oil.columns:
                        series = pd.to_numeric(oil[col], errors='coerce')
                        monthly = series.resample('ME').last().dropna()
                        results[name] = monthly
                        print(f"  ✓ {name:<30} {len(monthly)} 月")

            print(f"  CFTC 完成: {len(results)} 指标")
        else:
            print(f"  ✗ CFTC 下载失败: HTTP {resp.status_code}")

    except Exception as e:
        print(f"  ✗ CFTC 错误: {str(e)[:50]}")

    return pd.DataFrame(results) if results else pd.DataFrame()


# ═══════════════════════════════════════════════════════════
# 6. EIA API
# ═══════════════════════════════════════════════════════════

EIA_SERIES = {
    'us_oil_inventory_total': 'PET.WCESTUS1.W',
    'us_crude_production':    'PET.WCRFPUS2.W',
}

def fetch_eia():
    """从 EIA API 拉取能源数据。"""
    print(f"\n[EIA] 拉取 {len(EIA_SERIES)} 个指标...")
    results = {}

    for name, sid in EIA_SERIES.items():
        try:
            url = f"https://api.eia.gov/v2/seriesid/{sid}"
            resp = requests.get(url, params={'api_key': EIA_KEY}, timeout=30)
            if resp.status_code == 200:
                data = resp.json()
                if 'response' in data and 'data' in data['response']:
                    records = data['response']['data']
                    df_tmp = pd.DataFrame(records)
                    df_tmp['date'] = pd.to_datetime(df_tmp['period'])
                    df_tmp['value'] = pd.to_numeric(df_tmp['value'], errors='coerce')
                    series = df_tmp.set_index('date')['value'].sort_index()
                    monthly = series.resample('ME').last().dropna()
                    results[name] = monthly
                    print(f"  ✓ {name:<30} {len(monthly)} 月")
                else:
                    print(f"  ✗ {name:<30} 无数据")
            else:
                print(f"  ✗ {name:<30} HTTP {resp.status_code}")
        except Exception as e:
            print(f"  ✗ {name:<30} {str(e)[:40]}")

    return pd.DataFrame(results) if results else pd.DataFrame()


# ═══════════════════════════════════════════════════════════
# 7. GPR 地缘政治风险指数 (Caldara & Iacoviello)
# ═══════════════════════════════════════════════════════════

def fetch_gpr():
    """从学术网站自动下载 GPR 地缘政治风险指数。
    替代30个二值地缘事件虚拟变量。
    来源: https://www.matteoiacoviello.com/gpr.htm
    """
    print(f"\n[GPR] 下载地缘政治风险指数 (Caldara & Iacoviello)...")

    results = {}
    urls = [
        "https://www.matteoiacoviello.com/gpr_files/data_gpr_monthly.xls",
        "https://www.matteoiacoviello.com/gpr_files/data_gpr_daily_recent.xls",
    ]

    try:
        df_gpr = None
        for url in urls:
            try:
                df_gpr = pd.read_excel(url)
                print(f"  下载成功: {url.split('/')[-1]}")
                break
            except Exception:
                continue

        if df_gpr is not None and len(df_gpr) > 0:
            # 找到日期列和GPR列
            date_col = [c for c in df_gpr.columns if 'date' in c.lower() or 'month' in c.lower()]
            gpr_cols = [c for c in df_gpr.columns if 'gpr' in c.lower()]

            if date_col:
                df_gpr['date'] = pd.to_datetime(df_gpr[date_col[0]], errors='coerce')
            else:
                df_gpr['date'] = pd.to_datetime(df_gpr.iloc[:, 0], errors='coerce')

            df_gpr = df_gpr.dropna(subset=['date']).set_index('date').sort_index()

            for col in gpr_cols:
                series = pd.to_numeric(df_gpr[col], errors='coerce')
                monthly = series.resample('ME').last().dropna()
                if len(monthly) > 0:
                    clean_name = col.strip().lower().replace(' ', '_')
                    if clean_name in ('gpr', 'gprd'):
                        clean_name = 'gpr_index'
                    results[clean_name] = monthly
                    print(f"  ✓ {clean_name:<30} {len(monthly)} 月")

            if 'gpr_index' not in results and results:
                first_key = list(results.keys())[0]
                results['gpr_index'] = results[first_key]
                print(f"  ✓ gpr_index (alias of {first_key})")
        else:
            print(f"  ✗ GPR 数据为空")

    except Exception as e:
        print(f"  ✗ GPR 下载失败: {str(e)[:50]}")

    df = pd.DataFrame(results) if results else pd.DataFrame()
    print(f"  GPR 完成: {len(results)} 指标")
    return df


# ═══════════════════════════════════════════════════════════
# MERGE + FEATURE ENGINEERING
# ═══════════════════════════════════════════════════════════

def merge_all_sources(fred_df, yf_df, wb_df, ak_df, cftc_df, eia_df, gpr_df=None,
                      existing_path='output/panel_monthly.csv'):
    """合并所有数据源，生成更新后的面板。"""
    print(f"\n[MERGE] 合并所有数据源...")

    existing = pd.read_csv(existing_path, index_col=0, parse_dates=True)
    print(f"  现有面板: {len(existing)} 行 × {len(existing.columns)} 列")
    print(f"  截止: {existing.index[-1].strftime('%Y-%m')}")

    sources = [
        ('FRED', fred_df), ('YFinance', yf_df), ('WorldBank', wb_df),
        ('akshare', ak_df), ('CFTC', cftc_df), ('EIA', eia_df),
        ('GPR', gpr_df),
    ]

    updated_cols = 0
    new_rows = 0

    # 价格列需要用 FRED 数据覆盖已有值，确保数据源一致性
    PRICE_OVERRIDE_COLS = {'WTI_spot', 'Brent_spot'}

    for src_name, src_df in sources:
        if src_df is None or len(src_df) == 0:
            continue

        for col in src_df.columns:
            if col in existing.columns:
                is_price_override = (col in PRICE_OVERRIDE_COLS and src_name == 'FRED')
                # 补全缺失值 + 追加新日期
                for d in src_df.index:
                    if d not in existing.index:
                        existing.loc[d] = np.nan
                        new_rows += 1
                    if pd.notna(src_df.at[d, col]):
                        if pd.isna(existing.at[d, col]) or is_price_override:
                            existing.at[d, col] = src_df.at[d, col]
                            updated_cols += 1
            else:
                # 新列
                existing = existing.join(src_df[[col]], how='outer')

    existing = existing.sort_index()

    # 重新计算派生特征
    print(f"  重新计算派生特征...")
    try:
        from data_pipeline import engineer_features
        existing = engineer_features(existing)
    except Exception as e:
        print(f"  ⚠ engineer_features 跳过: {e}")

    # ISM 环比（22个）：从 ISM 子指数自动计算 MoM
    ism_mom_pairs = [
        ('美国:ISM:制造业PMI:环比',            '美国:ISM:制造业PMI'),
        ('美国:ISM:制造业PMI:新订单:环比',       '美国:ISM:制造业PMI:新订单'),
        ('美国:ISM:制造业PMI:订单库存:环比',      '美国:ISM:制造业PMI:自有库存'),
        ('美国:ISM:制造业PMI:自有库存:环比',      '美国:ISM:制造业PMI:自有库存'),
        ('美国:ISM:制造业PMI:客户库存:环比',      '美国:ISM:制造业PMI:客户库存'),
        ('美国:ISM:制造业PMI:新出口订单:环比',     '美国:ISM:制造业PMI:新出口订单'),
        ('美国:ISM:制造业PMI:进口:环比',         '美国:ISM:制造业PMI:进口'),
        ('美国:ISM:制造业PMI:产出:环比',         '美国:ISM:制造业PMI:产出'),
        ('美国:ISM:制造业PMI:供应商交付:环比',     '美国:ISM:制造业PMI:供应商交付'),
        ('美国:ISM:制造业PMI:物价:环比',         '美国:ISM:制造业PMI:物价'),
        ('美国:ISM:制造业PMI:就业:环比',         '美国:ISM:制造业PMI:就业'),
        ('美国:ISM:服务业PMI:环比',            '美国:ISM:服务业PMI'),
        ('美国:ISM:服务业PMI:商业活动:环比',      '美国:ISM:服务业PMI:商业活动'),
        ('美国:ISM:服务业PMI:新订单:环比',        '美国:ISM:服务业PMI:新订单'),
        ('美国:ISM:服务业PMI:就业:环比',         '美国:ISM:服务业PMI:就业'),
        ('美国:ISM:服务业PMI:供应商交付:环比',     '美国:ISM:服务业PMI:供应商交付'),
        ('美国:ISM:服务业PMI:库存:环比',         '美国:ISM:服务业PMI:库存'),
        ('美国:ISM:服务业PMI:物价:环比',         '美国:ISM:服务业PMI:物价'),
        ('美国:ISM:服务业PMI:订单库存:环比',      '美国:ISM:服务业PMI:订单库存'),
        ('美国:ISM:服务业PMI:新出口订单:环比',     '美国:ISM:服务业PMI:新出口订单'),
        ('美国:ISM:服务业PMI:进口:环比',         '美国:ISM:服务业PMI:进口'),
        ('美国:ISM:服务业PMI:库存景气:环比',      '美国:ISM:服务业PMI:库存'),
    ]
    ism_count = 0
    for mom_col, base_col in ism_mom_pairs:
        if base_col in existing.columns:
            existing[mom_col] = existing[base_col].diff()
            ism_count += 1
    if ism_count:
        print(f"  ✓ ISM 环比: {ism_count} 个派生特征")

    out_path = os.path.join(OUTPUT_DIR, 'panel_monthly_live.csv')
    existing.to_csv(out_path)
    print(f"  ✓ 更新后面板: {len(existing)} 行 × {len(existing.columns)} 列")
    print(f"  ✓ 截止: {existing.index[-1].strftime('%Y-%m')}")
    print(f"  ✓ 已保存: {out_path}")

    return existing


# ═══════════════════════════════════════════════════════════
# TEST MODE
# ═══════════════════════════════════════════════════════════

def test_apis():
    """测试各 API 连通性。"""
    print("=" * 65)
    print("API 连通性测试")
    print("=" * 65)

    results = {}

    # FRED
    try:
        from fredapi import Fred
        fred = Fred(api_key=FRED_KEY)
        d = fred.get_series('DCOILWTICO', observation_start='2026-01-01')
        results['FRED'] = f"✓ 连通 ({len(d)} 条)"
    except Exception as e:
        results['FRED'] = f"✗ {str(e)[:40]}"

    # YFinance
    try:
        import yfinance as yf
        d = yf.download('CL=F', period='5d', progress=False)
        results['YFinance'] = f"✓ 连通 ({len(d)} 条)"
    except Exception as e:
        results['YFinance'] = f"✗ {str(e)[:40]}"

    # World Bank
    try:
        resp = requests.head("https://thedocs.worldbank.org/en/doc/5d903e848db1d1b83e0ec8f744e55570-0350012021/related/CMO-Historical-Data-Monthly.xlsx",
                           timeout=10)
        results['WorldBank'] = f"✓ 可访问 (HTTP {resp.status_code})"
    except Exception as e:
        results['WorldBank'] = f"✗ {str(e)[:40]}"

    # akshare
    try:
        import akshare as ak
        results['akshare'] = f"✓ 已安装 (v{ak.__version__})"
    except Exception as e:
        results['akshare'] = f"✗ {str(e)[:40]}"

    # CFTC
    try:
        year = datetime.now().year
        resp = requests.head(f"https://www.cftc.gov/dea/newcot/{year}fut.zip", timeout=10)
        results['CFTC'] = f"✓ 可访问 (HTTP {resp.status_code})"
    except Exception as e:
        results['CFTC'] = f"✗ {str(e)[:40]}"

    # EIA
    try:
        resp = requests.get(f"https://api.eia.gov/v2/seriesid/PET.WCESTUS1.W",
                          params={'api_key': EIA_KEY}, timeout=10)
        results['EIA'] = f"✓ 连通 (HTTP {resp.status_code})"
    except Exception as e:
        results['EIA'] = f"✗ {str(e)[:40]}"

    # GPR
    try:
        resp = requests.head("https://www.matteoiacoviello.com/gpr_files/data_gpr_monthly.xls",
                           timeout=10)
        results['GPR'] = f"✓ 可访问 (HTTP {resp.status_code})"
    except Exception as e:
        results['GPR'] = f"✗ {str(e)[:40]}"

    print()
    for src, status in results.items():
        print(f"  {src:15s} {status}")

    ok = sum(1 for v in results.values() if '✓' in v)
    print(f"\n  {ok}/{len(results)} 个数据源可用")
    return results


# ═══════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════

def main(test=False, source=None):
    print("=" * 65)
    print("全特征 API 数据更新")
    print("=" * 65)

    if test:
        test_apis()
        return

    start = '2000-01-01'

    # 按数据源拉取
    fred_df = yf_df = wb_df = ak_df = cftc_df = eia_df = gpr_df = None

    if source is None or source == 'fred':
        fred_df = fetch_fred_bulk(start)

    if source is None or source == 'yfinance':
        yf_df = fetch_yfinance(start)

    if source is None or source == 'worldbank':
        wb_df = fetch_worldbank()

    if source is None or source == 'akshare':
        ak_df = fetch_akshare()

    if source is None or source == 'cftc':
        cftc_df = fetch_cftc()

    if source is None or source == 'eia':
        eia_df = fetch_eia()

    if source is None or source == 'gpr':
        gpr_df = fetch_gpr()

    # 合并
    panel = merge_all_sources(fred_df, yf_df, wb_df, ak_df, cftc_df, eia_df, gpr_df)

    print(f"\n{'='*65}")
    print("✅ 数据更新完成")
    print(f"{'='*65}")


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='全特征 API 数据更新')
    parser.add_argument('--test', action='store_true', help='测试 API 连通性')
    parser.add_argument('--source', choices=['fred', 'yfinance', 'worldbank',
                                             'akshare', 'cftc', 'eia', 'gpr'],
                        help='只更新某个数据源')
    args = parser.parse_args()
    main(test=args.test, source=args.source)