Spaces:

Entelechele
/

Automatic_Stock

Sleeping

File size: 15,658 Bytes

b1c2e5f

# 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 = """
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8"/>
<meta name="viewport" content="width=device-width,initial-scale=1"/>
<title>指数估值分位报告（pyecharts + FastAPI，离线，宽表版）</title>
<style>
body{font-family:system-ui,-apple-system,Segoe UI,Roboto,Helvetica,Arial,'Noto Sans',sans-serif;margin:16px;}
h1{margin:0 0 12px;font-size:20px;}
.controls{display:grid;gap:8px;grid-template-columns:repeat(4,minmax(160px,1fr));margin-bottom:12px;}
.control label{display:block;font-size:12px;color:#555;margin-bottom:4px;}
select{width:100%;padding:6px;border-radius:8px;border:1px solid #ddd;}
#chart{width:100%;height:560px;}
table{width:100%;border-collapse:collapse;margin-top:10px;}
th,td{border:1px solid #e6e6e6;padding:6px 8px;text-align:right;}
th:first-child,td:first-child{text-align:left;}
.muted{color:#777;font-size:12px;}
</style>
</head>
<body>
<h1>指数估值分位报告（pyecharts + FastAPI，离线，宽表版）</h1>

<div class="controls">
  <div class="control"><label>指数 (Index)</label><select id="selIndex"></select></div>
  <div class="control"><label>估值指标 (Metric)</label><select id="selMetric"></select></div>
  <div class="control"><label>计算方法 (Method)</label><select id="selMethod"></select></div>
  <div class="control"><label>历史窗口 (Window)</label>
    <select id="selWindow"><option>3Y</option><option selected>10Y</option><option>Expanding</option></select>
  </div>
</div>

<div id="chart"></div>

<table id="summary">
  <thead><tr>
    <th>最新交易日</th><th>当前指标值</th>
    <th>3Y分位(%)</th><th>10Y分位(%)</th><th>Expanding分位(%)</th>
    <th>μ</th><th>μ-1σ</th><th>μ+1σ</th><th>μ-2σ</th><th>μ+2σ</th><th>μ-3σ</th><th>μ+3σ</th>
  </tr></thead>
  <tbody></tbody>
</table>

<p class="muted">完全离线：前端使用本地 /static/echarts.min.js；图表由后端 pyecharts 生成。</p>

<script src="/static/echarts.min.js"></script>
<script>
const selIndex  = document.getElementById('selIndex');
const selMetric = document.getElementById('selMetric');
const selMethod = document.getElementById('selMethod');
const selWindow = document.getElementById('selWindow');

async function jget(url){ const r = await fetch(url); return await r.json(); }
async function getText(url){ const r = await fetch(url); return await r.text(); }

function injectHTMLAndRunScripts(container, html){
  container.innerHTML = html;

  // Collect scripts as array to avoid live NodeList issues
  const scripts = Array.from(container.querySelectorAll('script'));

  scripts.forEach(oldS => {
    try {
      const parent = oldS.parentNode;
      const s = document.createElement('script');

      // Copy attributes (type/src/async/etc.)
      Array.from(oldS.attributes || []).forEach(attr => {
        try {
          s.setAttribute(attr.name, attr.value);
        } catch(e) {
          // 某些 attribute 可能不可复制，记录但不阻塞
          console.warn('Failed to copy attribute', attr.name, attr.value, e);
        }
      });

      if (oldS.src) {
        // External script: append to head to ensure browser loads & executes it
        // Use async=false to request execution order (best-effort)
        s.src = oldS.src;
        try { s.async = false; } catch(e){}
        // Append and then remove the old node (if still attached)
        document.head.appendChild(s);
        if (parent && parent.contains(oldS)) {
          try { parent.removeChild(oldS); } catch(e){ /* ignore */ }
        }
      } else {
        // Inline script: copy text and replace if possible
        const code = oldS.textContent || oldS.innerHTML || '';
        s.text = code;
        if (parent && parent.contains(oldS)) {
          try {
            parent.replaceChild(s, oldS);
          } catch (e) {
            // fallback: append to container
            console.warn('replaceChild failed for inline script, appending instead', e);
            container.appendChild(s);
            try { parent.removeChild(oldS); } catch(e){}
          }
        } else {
          // parent missing (script detached) -> just append
          container.appendChild(s);
        }
      }
    } catch (err) {
      // 打印出错信息并尽量继续处理其它脚本
      console.error('injectHTMLAndRunScripts error while handling a script node:', err);
      try {
        console.log('Problematic script:', oldS && (oldS.src || oldS.textContent || oldS.innerHTML));
      } catch(e){ console.warn('Could not read script content:', e); }
    }
  });
}

async function init(){
  const meta = await jget('/meta');
  selIndex.innerHTML  = meta.indexes.map(v=>`<option value="${v}">${v}</option>`).join('');
  selMetric.innerHTML = meta.metrics.map(v=>`<option value="${v}">${v}</option>`).join('');
  selMethod.innerHTML = meta.methods.map(v=>`<option value="${v}">${v}</option>`).join('');
  await render();
}

async function render(){
  const params = new URLSearchParams({
    index_code: selIndex.value, metric: selMetric.value, method: selMethod.value, window: selWindow.value
  });
  const chartHTML = await getText('/chart?' + params.toString());
  injectHTMLAndRunScripts(document.getElementById('chart'), chartHTML);

  const summary = await jget('/summary?' + params.toString());
  const tbody = document.querySelector('#summary tbody');
  const fmt = n => (n==null || isNaN(n)) ? '' : Number(n).toFixed(4);
  tbody.innerHTML = `<tr>
    <td style="text-align:left">${summary.date || ''}</td>
    <td>${fmt(summary.value)}</td>
    <td>${fmt(summary.pct_3Y)}</td>
    <td>${fmt(summary.pct_10Y)}</td>
    <td>${fmt(summary.pct_Expanding)}</td>
    <td>${fmt(summary.mu)}</td>
    <td>${fmt(summary.n1)}</td>
    <td>${fmt(summary.p1)}</td>
    <td>${fmt(summary.n2)}</td>
    <td>${fmt(summary.p2)}</td>
    <td>${fmt(summary.n3)}</td>
    <td>${fmt(summary.p3)}</td>
  </tr>`;
}

selIndex.addEventListener('change', render);
selMetric.addEventListener('change', render);
selMethod.addEventListener('change', render);
selWindow.addEventListener('change', render);

init();
</script>
</body>
</html>
"""
    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("<div style='padding:12px;color:#b00'>未找到对应列</div>")

    g = df[df[INDEX_COL] == index_code].sort_values(DATE_COL)
    if g.empty:
        return HTMLResponse("<div style='padding:12px;color:#b00'>无该指数数据</div>")

    # 根据窗口类型过滤数据，显示不同的起始点
    mu_col = f"{base_col}_mu_{window}"
    if mu_col not in g.columns:
        return HTMLResponse("<div style='padding:12px;color:#b00'>无该窗口数据</div>")
    
    # 找到第一个非NaN的数据点作为起始点
    first_valid_idx = g[mu_col].first_valid_index()
    if first_valid_idx is None:
        return HTMLResponse("<div style='padding:12px;color:#b00'>该窗口无有效数据</div>")
    
    # 从第一个有效数据点开始
    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 + '<br/>值: ' + (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"""
    <!DOCTYPE html>
    <html>
    <head>
    <meta charset="utf-8">
    <title>{index_code} {metric} 图表</title>
    <script>{echarts_js}</script>
    </head>
    <body>
    {line.render_embed()}
    </body>
    </html>
    """
    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}"),
    })