app.py

from __future__ import annotations

from datetime import date
from pathlib import Path

import pandas as pd
import streamlit as st

from quant_wind.data_provider import DEFAULT_UNIVERSE_PATH, WindDataError, WindDataProvider, generate_sample_market_data, read_universe
from quant_wind.reporting import export_result
from quant_wind.strategy import BacktestConfig, run_rotation_backtest


OUTPUT_DIR = Path("outputs")


def main() -> None:
    st.set_page_config(page_title="行业 ETF 轮动", layout="wide")
    st.title("行业 ETF 轮动策略看板")

    params = sidebar_params()
    if not params["run"]:
        st.info("设置参数后点击“运行回测”。")
        return

    try:
        universe = read_universe(params["universe_path"])
        market_data = load_market_data(params, universe)
        config = BacktestConfig(
            start_date=params["start_date"].isoformat(),
            end_date=params["end_date"].isoformat(),
            holding_count=params["holding_count"],
            min_holding_count=params["min_holding_count"],
            rebalance_frequency=params["rebalance_frequency"],
            momentum_windows=params["momentum_windows"],
            volatility_window=params["volatility_window"],
            moving_average_window=params["moving_average_window"],
            liquidity_window=params["liquidity_window"],
            min_history_days=params["min_history_days"],
            min_avg_amount=params["min_avg_amount_million"] * 1_000_000,
            volatility_penalty=params["volatility_penalty"],
            transaction_cost=params["transaction_cost_bps"] / 10_000,
        )
        result = run_rotation_backtest(market_data.prices, market_data.amounts, market_data.names, config)
        exported = export_result(result, OUTPUT_DIR)
    except (ValueError, FileNotFoundError, WindDataError) as exc:
        st.error(str(exc))
        return

    st.caption(f"数据源：{market_data.source} | 标的数：{len(market_data.prices.columns)} | 导出目录：{OUTPUT_DIR.resolve()}")
    render_metrics(result.metrics)

    tab_perf, tab_holdings, tab_latest, tab_exports, tab_data = st.tabs(
        ["绩效", "持仓明细", "最新信号", "导出文件", "数据检查"]
    )
    with tab_perf:
        render_performance(result.daily)
    with tab_holdings:
        render_table("调仓信号", result.signals, height=420)
        render_table("交易变化", result.trades, height=300)
    with tab_latest:
        render_table("下一期目标持仓", result.latest_signal, height=320)
    with tab_exports:
        render_exports(exported)
    with tab_data:
        render_data_checks(market_data.prices, market_data.amounts, universe)


def sidebar_params() -> dict:
    st.sidebar.header("参数")
    today = date.today()
    default_start = date(today.year - 3, 1, 1)

    data_source = st.sidebar.radio("数据源", ["示例数据", "WindPy"], horizontal=True)
    start_date = st.sidebar.date_input("回测开始", value=default_start)
    end_date = st.sidebar.date_input("回测结束", value=today)
    universe_path = st.sidebar.text_input("ETF 标的池 CSV", value=str(DEFAULT_UNIVERSE_PATH))

    holding_count = st.sidebar.number_input("持仓数量", min_value=1, max_value=20, value=5, step=1)
    min_holding_count = st.sidebar.number_input("最低满仓标的数", min_value=1, max_value=20, value=3, step=1)
    rebalance_label = st.sidebar.selectbox("调仓频率", ["周频", "月频", "日频"], index=0)
    rebalance_frequency = {"周频": "W-FRI", "月频": "M", "日频": "D"}[rebalance_label]
    raw_momentum_windows = st.sidebar.text_input("动量窗口", value="20,60,120")
    try:
        momentum_windows = parse_windows(raw_momentum_windows)
    except ValueError as exc:
        st.sidebar.error(str(exc))
        momentum_windows = (20, 60, 120)
        run_disabled = True
    else:
        run_disabled = False

    st.sidebar.subheader("风控与成本")
    moving_average_window = st.sidebar.number_input("趋势均线窗口", min_value=20, max_value=260, value=120, step=5)
    volatility_window = st.sidebar.number_input("波动率窗口", min_value=10, max_value=260, value=60, step=5)
    liquidity_window = st.sidebar.number_input("流动性窗口", min_value=5, max_value=120, value=20, step=5)
    min_history_days = st.sidebar.number_input("最短历史天数", min_value=20, max_value=260, value=120, step=5)
    min_avg_amount_million = st.sidebar.number_input("最低日均成交额(百万元)", min_value=0.0, value=30.0, step=5.0)
    volatility_penalty = st.sidebar.slider("波动惩罚", min_value=0.0, max_value=2.0, value=0.35, step=0.05)
    transaction_cost_bps = st.sidebar.number_input("单边成本(bp)", min_value=0.0, value=12.5, step=0.5)

    run = st.sidebar.button("运行回测", type="primary", width="stretch", disabled=run_disabled)
    return {
        "data_source": data_source,
        "start_date": start_date,
        "end_date": end_date,
        "universe_path": universe_path,
        "holding_count": int(holding_count),
        "min_holding_count": int(min_holding_count),
        "rebalance_frequency": rebalance_frequency,
        "momentum_windows": momentum_windows,
        "moving_average_window": int(moving_average_window),
        "volatility_window": int(volatility_window),
        "liquidity_window": int(liquidity_window),
        "min_history_days": int(min_history_days),
        "min_avg_amount_million": float(min_avg_amount_million),
        "volatility_penalty": float(volatility_penalty),
        "transaction_cost_bps": float(transaction_cost_bps),
        "run": run,
    }


def parse_windows(raw: str) -> tuple[int, ...]:
    windows = tuple(int(part.strip()) for part in raw.split(",") if part.strip())
    if not windows:
        raise ValueError("动量窗口不能为空。")
    return windows


@st.cache_data(show_spinner=False)
def load_market_data(params: dict, universe: pd.DataFrame):
    warmup_start = pd.Timestamp(params["start_date"]) - pd.tseries.offsets.BDay(320)
    if params["data_source"] == "WindPy":
        provider = WindDataProvider(params["universe_path"])
        return provider.fetch_market_data(params["start_date"], params["end_date"], warmup_days=320)
    return generate_sample_market_data(universe, warmup_start, params["end_date"])


def render_metrics(metrics: pd.Series) -> None:
    labels = [
        ("累计收益", "total_return", "percent"),
        ("年化收益", "annual_return", "percent"),
        ("最大回撤", "max_drawdown", "percent"),
        ("年化波动", "volatility", "percent"),
        ("夏普", "sharpe", "number"),
        ("卡玛", "calmar", "number"),
        ("月胜率", "monthly_win_rate", "percent"),
        ("年化换手", "annual_turnover", "number"),
    ]
    cols = st.columns(4)
    for idx, (label, key, kind) in enumerate(labels):
        value = metrics.get(key, float("nan"))
        cols[idx % 4].metric(label, format_value(value, kind))


def render_performance(daily: pd.DataFrame) -> None:
    left, right = st.columns(2)
    with left:
        st.subheader("净值曲线")
        st.line_chart(daily[["nav"]])
    with right:
        st.subheader("回撤")
        st.area_chart(daily[["drawdown"]])

    st.subheader("日度数据")
    st.dataframe(format_display_frame(daily.tail(500)), width="stretch", height=320)


def render_table(title: str, frame: pd.DataFrame, height: int) -> None:
    st.subheader(title)
    if frame.empty:
        st.info("暂无数据。")
        return
    st.dataframe(format_display_frame(frame.tail(500)), width="stretch", height=height)


def render_exports(paths: dict[str, Path]) -> None:
    for name, path in paths.items():
        exists = path.exists()
        st.write(f"{name}: `{path.resolve()}`" if exists else f"{name}: 未生成")


def render_data_checks(prices: pd.DataFrame, amounts: pd.DataFrame, universe: pd.DataFrame) -> None:
    checks = pd.DataFrame(
        {
            "code": prices.columns,
            "name": [dict(zip(universe["code"], universe["name"])).get(code, code) for code in prices.columns],
            "first_price_date": [prices[code].first_valid_index() for code in prices.columns],
            "last_price_date": [prices[code].last_valid_index() for code in prices.columns],
            "missing_price_ratio": prices.isna().mean().values,
            "avg_amount": amounts.mean().values,
        }
    )
    st.dataframe(format_display_frame(checks), width="stretch", height=520)


def format_display_frame(frame: pd.DataFrame) -> pd.DataFrame:
    display = frame.copy()
    for col in display.columns:
        if pd.api.types.is_datetime64_any_dtype(display[col]):
            display[col] = display[col].dt.strftime("%Y-%m-%d")
    return display


def format_value(value: float, kind: str) -> str:
    if pd.isna(value):
        return "-"
    if kind == "percent":
        return f"{value:.2%}"
    return f"{value:.2f}"


if __name__ == "__main__":
    main()