#!/usr/bin/env python3
"""Fetch public monthly macro and global-cycle controls used by the report.

FRED is the distribution channel. The output preserves the FRED series ID,
underlying provider and proxy boundary instead of presenting every series as
an original Chinese official release.
"""

from __future__ import annotations

import csv
import io
import json
import math
import urllib.request
from pathlib import Path

import pandas as pd


ROOT = Path(__file__).resolve().parents[1]
RAW = ROOT / "data" / "research" / "raw" / "macro"
OUT = ROOT / "data" / "research" / "derived"
START = pd.Timestamp("1990-01-01")
END = pd.Timestamp("2025-12-31")

SERIES = {
    "中国M2": {
        "id": "MYAGM2CNM189N",
        "provider": "国际货币基金组织（FRED分发）",
        "unit": "本币",
        "role": "国内信用与流动性",
        "transform": "同比增速",
        "boundary": "非中国人民银行原始下载表；用于共同周期控制",
    },
    "中国3个月同业利率": {
        "id": "IR3TIB01CNM156N",
        "provider": "OECD（FRED分发）",
        "unit": "%",
        "role": "国内短端利率",
        "transform": "水平值",
        "boundary": "不是贷款基准利率或LPR；只代表短端市场利率",
    },
    "中国PPI": {
        "id": "CHNPIEATI01GYM",
        "provider": "OECD（FRED分发）",
        "unit": "同比%",
        "role": "国内工业价格周期",
        "transform": "原值",
        "boundary": "用于月度控制；年度表仍以国家统计局口径为主",
    },
    "人民币兑美元": {
        "id": "EXCHUS",
        "provider": "美国联邦储备委员会（FRED分发）",
        "unit": "人民币/美元",
        "role": "汇率与出口条件",
        "transform": "月均值及月变化率",
        "boundary": "市场汇率代理，不代替政策中间价制度分析",
    },
    "全球科技周期": {
        "id": "NASDAQCOM",
        "provider": "Nasdaq（FRED分发）",
        "unit": "指数点",
        "role": "全球科技风险偏好代理",
        "transform": "月末值及月收益",
        "boundary": "纳斯达克综合价格指数，不是全球科技总收益指数",
    },
    "全球商品周期": {
        "id": "PALLFNFINDEXM",
        "provider": "国际货币基金组织（FRED分发）",
        "unit": "2016=100",
        "role": "全球大宗商品价格代理",
        "transform": "月度指数及月变化率",
        "boundary": "综合商品指数，不能代表单一煤炭、钢铁或油价",
    },
}


def download_series(series_id: str) -> str:
    url = f"https://fred.stlouisfed.org/graph/fredgraph.csv?id={series_id}"
    request = urllib.request.Request(url, headers={"User-Agent": "A-share-research/1.0"})
    with urllib.request.urlopen(request, timeout=60) as response:
        return response.read().decode("utf-8-sig")


def parse_series(text: str, series_id: str) -> pd.DataFrame:
    frame = pd.read_csv(io.StringIO(text))
    frame.columns = ["date", "value"]
    frame["date"] = pd.to_datetime(frame["date"], errors="coerce")
    frame["value"] = pd.to_numeric(frame["value"], errors="coerce")
    frame = frame[frame["date"].between(START, END)].dropna(subset=["date"])
    frame["series_id"] = series_id
    return frame


def to_monthly(frame: pd.DataFrame) -> pd.DataFrame:
    values = frame.set_index("date")["value"].sort_index()
    monthly = values.resample("ME").last()
    return monthly.rename("level").reset_index()


def main() -> None:
    RAW.mkdir(parents=True, exist_ok=True)
    OUT.mkdir(parents=True, exist_ok=True)
    long_rows: list[pd.DataFrame] = []
    metadata: list[dict] = []
    for name, definition in SERIES.items():
        raw_path = RAW / f"fred_{definition['id']}.csv"
        if raw_path.exists():
            text = raw_path.read_text(encoding="utf-8-sig")
        else:
            text = download_series(definition["id"])
            raw_path.write_text(text, encoding="utf-8")
        parsed = parse_series(text, definition["id"])
        monthly = to_monthly(parsed)
        monthly["指标"] = name
        monthly["月变化率"] = monthly["level"].pct_change(fill_method=None)
        monthly["同比变化率"] = monthly["level"].pct_change(12, fill_method=None)
        monthly["单位"] = definition["unit"]
        monthly["角色"] = definition["role"]
        monthly["FRED序列"] = definition["id"]
        monthly["底层提供方"] = definition["provider"]
        monthly["代理边界"] = definition["boundary"]
        monthly = monthly.rename(columns={"date": "月份", "level": "数值"})
        long_rows.append(monthly)
        metadata.append(
            {
                "指标": name,
                "FRED序列": definition["id"],
                "底层提供方": definition["provider"],
                "转换": definition["transform"],
                "有效起始月": monthly.loc[monthly["数值"].notna(), "月份"].min(),
                "有效结束月": monthly.loc[monthly["数值"].notna(), "月份"].max(),
                "有效月数": int(monthly["数值"].notna().sum()),
                "代理边界": definition["boundary"],
                "原始链接": f"https://fred.stlouisfed.org/series/{definition['id']}",
            }
        )

    panel = pd.concat(long_rows, ignore_index=True)
    panel["缺失标记"] = panel["数值"].isna().map({True: "是", False: "否"})
    panel.to_csv(OUT / "macro_controls_monthly.csv", index=False, encoding="utf-8-sig")
    pd.DataFrame(metadata).to_csv(
        OUT / "macro_controls_monthly_metadata.csv", index=False, encoding="utf-8-sig"
    )

    available = panel.groupby("指标")["数值"].apply(lambda values: int(values.notna().sum()))
    assert set(available.index) == set(SERIES)
    assert (available > 0).all()
    assert panel["月份"].between(START, END).all()
    assert panel["缺失标记"].isin(["是", "否"]).all()
    print("monthly-macro-control assertions: OK")


if __name__ == "__main__":
    main()