Spaces:

Entelechele
/

Automatic_Stock

Sleeping

App Files Files Community

Entelechele commited on Oct 14

Commit

b1c2e5f

verified ·

1 Parent(s): 7e2e2b9

Create app.py

Browse files

Files changed (1) hide show

app.py +416 -0

app.py ADDED Viewed

	@@ -0,0 +1,416 @@

+# 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}"),
+    })