完善知行短线

backend/app/data/tushare_provider.py

@@ -15,14 +15,29 @@ def _parse(date_str: str) -> datetime:
     return datetime.strptime(str(date_str), "%Y%m%d")
 
 
+def get_pro():
+    """创建 pro 客户端（统一入口）。
+
+    15000 积分档 token 只认 quicksync 镜像（直连 api.tushare.pro 会 40101 token 不对），
+    官方接入方式是改写 SDK 的私有类属性 DataApi.__http_url——类属性是所有实例
+    （含 pro_bar 内部自建的 DataApi）共用的请求地址，打一次补丁全局生效。
+    """
+    import tushare as ts
+    import tushare.pro.client as client
+
+    if not settings.tushare_token:
+        raise RuntimeError("未配置 TUSHARE_TOKEN")
+    if settings.tushare_api_url:
+        client.DataApi._DataApi__http_url = settings.tushare_api_url
+    ts.set_token(settings.tushare_token)
+    return ts.pro_api()
+
+
 def fetch_daily(code: str, start: str = "20200101", end: str | None = None,
                 adjust: str = "qfq") -> list[Bar]:
     import tushare as ts  # 延迟导入：未装或无 token 时该数据源不可用
 
-    if not settings.tushare_token:
-        raise RuntimeError("未配置 TUSHARE_TOKEN")
-    ts.set_token(settings.tushare_token)
-    pro = ts.pro_api()
+    pro = get_pro()  # 含镜像补丁（quicksync）+ token 检查
     ts_code = to_ts_code(code)
     end = end or datetime.now().strftime("%Y%m%d")
 
@@ -51,3 +66,32 @@ def fetch_daily(code: str, start: str = "20200101", end: str | None = None,
             )
         )
     return bars
+
+
+def fetch_chips(ts_code: str, end: str | None = None):
+    """拉取筹码分布截面：cyq_perf（成本分位/获利比例/平均成本）+ cyq_chips（价位→占比）。
+
+    end 为 YYYYMMDD 参考日：取 <=end 的最近有数据交易日（周/月 K 线由前端换算成
+    周期末传入，这里只需向前吸附到实际数据日）；None 取最新。
+    数据自 2018 年起，早于此返回 (None, [])。返回 (perf_dict|None, [(price, percent)])。
+    """
+    from datetime import timedelta
+
+    pro = get_pro()
+    e = end or datetime.now().strftime("%Y%m%d")
+    if e < "20180101":
+        return None, []
+    # 吸附余量 60 个自然日：覆盖春节等长假 + 月初参考日（如月 K 传上月末）
+    s = (datetime.strptime(e, "%Y%m%d") - timedelta(days=60)).strftime("%Y%m%d")
+    s = max(s, "20180101")
+
+    perf = pro.cyq_perf(ts_code=ts_code, start_date=s, end_date=e)
+    if perf is None or perf.empty:
+        return None, []
+    perf = perf.sort_values("trade_date").iloc[-1]  # <=end 的最近一条
+    d = str(perf["trade_date"])
+    chips = pro.cyq_chips(ts_code=ts_code, trade_date=d)
+    rows: list[tuple[float, float]] = []
+    if chips is not None and not chips.empty:
+        rows = [(float(p), float(v)) for p, v in zip(chips["price"], chips["percent"])]
+    return perf.to_dict(), rows