# app.py
# -*- coding: utf-8 -*-
from __future__ import annotations
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import re
import pandas as pd
from fastapi import FastAPI, Query
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from pyecharts import options as opts
from pyecharts.charts import Line
from pyecharts.globals import CurrentConfig
from pyecharts.commons.utils import JsCode # <-- 关键：用于包装 JS 函数
# ---------------- 0) 路径 ----------------
DATA_DIR = Path(".")
WIDE_PATH = DATA_DIR / "bands_df.parquet" # 你的“宽表+追加bands”文件
DATE_COL = "trade_date"
INDEX_COL = "index_code"
# ---------------- 1) 读数 ----------------
if not WIDE_PATH.exists():
raise FileNotFoundError(
f"未找到 {WIDE_PATH}。请先保存：bands_appended.to_parquet('bands_appended.parquet')"
)
df = pd.read_parquet(WIDE_PATH).copy()
if DATE_COL not in df.columns or INDEX_COL not in df.columns:
raise ValueError(f"数据必须包含 {DATE_COL} 与 {INDEX_COL} 列。")
df[DATE_COL] = pd.to_datetime(df[DATE_COL])
df = df.sort_values([INDEX_COL, DATE_COL]).reset_index(drop=True)
df["date_str"] = df[DATE_COL].dt.strftime("%Y-%m-%d")
# ---------------- 2) 识别基础列 ----------------
def is_base_col(c: str) -> bool:
if c in (DATE_COL, INDEX_COL): return False
if not pd.api.types.is_numeric_dtype(df[c]): return False
return not any(tok in c for tok in ("_mu_", "_sigma_", "_band_", "_pct_"))
BASE_PATTERN = re.compile(r"^(pe_ttm|pb_lf|ps_ttm|pcf_ttm)_(weighted|median|total)$", re.I)
METRIC_MAP = {"pe_ttm": "PE-TTM", "pb_lf": "PB-LF", "ps_ttm": "PS-TTM", "pcf_ttm": "PCF-TTM"}
METHOD_MAP = {"weighted": "权重加权法", "median": "中值法", "total": "总量法"}
base_cols: List[str] = [c for c in df.columns if is_base_col(c)]
KEY_TO_BASE: Dict[Tuple[str, str], str] = {}
for c in base_cols:
m = BASE_PATTERN.match(c)
if not m: continue
metric_key, method_key = m.group(1).lower(), m.group(2).lower()
KEY_TO_BASE[(METRIC_MAP.get(metric_key, metric_key),
METHOD_MAP.get(method_key, method_key))] = c
INDEXES = sorted(df[INDEX_COL].unique().tolist())
METRICS = sorted({k[0] for k in KEY_TO_BASE})
METHODS = sorted({k[1] for k in KEY_TO_BASE})
WINDOWS = ["3Y", "10Y", "Expanding"]
# ---------------- 3) FastAPI ----------------
app = FastAPI(title="Valuation (pyecharts + FastAPI, offline, wide)")
static_dir = Path("./static")
static_dir.mkdir(exist_ok=True)
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
# ---------------- 4) 首页 ----------------
@app.get("/", response_class=HTMLResponse)
def index() -> str:
html = """
指数估值分位报告（pyecharts + FastAPI，离线，宽表版）
指数估值分位报告（pyecharts + FastAPI，离线，宽表版）
| 最新交易日 | 当前指标值 |
3Y分位(%) | 10Y分位(%) | Expanding分位(%) |
μ | μ-1σ | μ+1σ | μ-2σ | μ+2σ | μ-3σ | μ+3σ |
完全离线：前端使用本地 /static/echarts.min.js；图表由后端 pyecharts 生成。
"""
return HTMLResponse(content=html)
# ---------------- 5) 元数据 ----------------
@app.get("/meta")
def meta() -> JSONResponse:
return JSONResponse({"indexes": INDEXES, "metrics": METRICS, "methods": METHODS, "windows": WINDOWS})
# ---------------- 6) 图表 ----------------
@app.get("/chart", response_class=HTMLResponse)
def chart_offline(
index_code: str = Query(...),
metric: str = Query(...),
method: str = Query(...),
window: str = Query(..., pattern="^(3Y|10Y|Expanding)$")
) -> str:
import json
base_col = KEY_TO_BASE.get((metric, method))
if base_col is None:
return HTMLResponse("未找到对应列
")
g = df[df[INDEX_COL] == index_code].sort_values(DATE_COL)
if g.empty:
return HTMLResponse("无该指数数据
")
# 根据窗口类型过滤数据，显示不同的起始点
mu_col = f"{base_col}_mu_{window}"
if mu_col not in g.columns:
return HTMLResponse("无该窗口数据
")
# 找到第一个非NaN的数据点作为起始点
first_valid_idx = g[mu_col].first_valid_index()
if first_valid_idx is None:
return HTMLResponse("该窗口无有效数据
")
# 从第一个有效数据点开始
g_filtered = g.loc[first_valid_idx:].copy()
dates = g_filtered["date_str"].tolist()
y_value = g_filtered[base_col].tolist()
def col_or_nan(c):
if c in g_filtered.columns:
return g_filtered[c].tolist()
return [None] * len(g_filtered)
mu = col_or_nan(f"{base_col}_mu_{window}")
p1 = col_or_nan(f"{base_col}_band_p1_{window}")
n1 = col_or_nan(f"{base_col}_band_n1_{window}")
p2 = col_or_nan(f"{base_col}_band_p2_{window}")
n2 = col_or_nan(f"{base_col}_band_n2_{window}")
p3 = col_or_nan(f"{base_col}_band_p3_{window}")
n3 = col_or_nan(f"{base_col}_band_n3_{window}")
# 构建图表
line = (
Line(init_opts=opts.InitOpts(width="100%", height="560px"))
.add_xaxis(dates)
.add_yaxis(
"主线",
y_value,
is_symbol_show=False,
label_opts=opts.LabelOpts(is_show=False), # 不显示标签
tooltip_opts=opts.TooltipOpts( # 自定义tooltip
trigger="axis",
formatter=JsCode("""
function(params) {
var date = params[0].axisValue;
var value = params[0].value;
return '日期: ' + date + '
值: ' + (value ? value.toFixed(4) : 'N/A');
}
""")
)
)
.add_yaxis(
"μ",
mu,
is_symbol_show=False,
linestyle_opts=opts.LineStyleOpts(width=1),
tooltip_opts=opts.TooltipOpts(is_show=False) # 其他线不显示tooltip
)
.add_yaxis(
"μ+1σ",
p1,
is_symbol_show=False,
linestyle_opts=opts.LineStyleOpts(width=1),
tooltip_opts=opts.TooltipOpts(is_show=False)
)
.add_yaxis(
"μ-1σ",
n1,
is_symbol_show=False,
linestyle_opts=opts.LineStyleOpts(width=1),
tooltip_opts=opts.TooltipOpts(is_show=False)
)
.add_yaxis(
"μ+2σ",
p2,
is_symbol_show=False,
linestyle_opts=opts.LineStyleOpts(width=1),
tooltip_opts=opts.TooltipOpts(is_show=False)
)
.add_yaxis(
"μ-2σ",
n2,
is_symbol_show=False,
linestyle_opts=opts.LineStyleOpts(width=1),
tooltip_opts=opts.TooltipOpts(is_show=False)
)
.add_yaxis(
"μ+3σ",
p3,
is_symbol_show=False,
linestyle_opts=opts.LineStyleOpts(width=1),
tooltip_opts=opts.TooltipOpts(is_show=False)
)
.add_yaxis(
"μ-3σ",
n3,
is_symbol_show=False,
linestyle_opts=opts.LineStyleOpts(width=1),
tooltip_opts=opts.TooltipOpts(is_show=False)
)
.set_global_opts(
title_opts=opts.TitleOpts(title=f"{index_code} {metric} ({method}) - {window} 历史分位图"),
legend_opts=opts.LegendOpts(pos_top="5%"),
xaxis_opts=opts.AxisOpts(type_="category", name="日期"),
yaxis_opts=opts.AxisOpts(type_="value", name=metric),
datazoom_opts=[opts.DataZoomOpts(type_="inside"), opts.DataZoomOpts()],
tooltip_opts=opts.TooltipOpts(
trigger="axis",
axis_pointer_type="cross"
)
)
)
# 读取 echarts.min.js 内容
echarts_js = (Path("./static/echarts.min.js")).read_text(encoding="utf-8")
# 生成完整 HTML
full_html = f"""
{index_code} {metric} 图表
{line.render_embed()}
"""
return HTMLResponse(full_html)
# ---------------- 7) 摘要 ----------------
@app.get("/summary")
def summary(index_code: str = Query(...), metric: str = Query(...), method: str = Query(...),
window: str = Query(..., regex="^(3Y|10Y|Expanding)$")) -> JSONResponse:
base_col = KEY_TO_BASE.get((metric, method))
if base_col is None:
return JSONResponse({"error": "bad metric/method"}, status_code=400)
g = df[df[INDEX_COL] == index_code].copy()
if g.empty: return JSONResponse({"error": "no data"}, status_code=404)
last_date = g[DATE_COL].max()
row = g[g[DATE_COL] == last_date].iloc[0]
def get_safe(c):
if c in g.columns:
v = row[c]
try: return None if pd.isna(v) else float(v)
except: return None
return None
return JSONResponse({
"date": last_date.strftime("%Y-%m-%d"),
"value": get_safe(base_col),
"pct_3Y": get_safe(f"{base_col}_pct_3Y"),
"pct_10Y": get_safe(f"{base_col}_pct_10Y"),
"pct_Expanding": get_safe(f"{base_col}_pct_Expanding"),
"mu": get_safe(f"{base_col}_band_mu_{window}"),
"n1": get_safe(f"{base_col}_band_n1_{window}"),
"p1": get_safe(f"{base_col}_band_p1_{window}"),
"n2": get_safe(f"{base_col}_band_n2_{window}"),
"n3": get_safe(f"{base_col}_band_n3_{window}"),
"p2": get_safe(f"{base_col}_band_p2_{window}"),
"p3": get_safe(f"{base_col}_band_p3_{window}"),
})