tradingSystem/app/services/publish_export.py

# -*- coding: utf-8 -*-
"""
公示表导出 (每日持仓 / 净值 → 公司《量化数据》xlsx)
====================================================
公司流程: 每日把持仓与净值导出为 excel 公示。本模块按公司模板
(量化数据2026.8.3.xlsx) 的版式与口径生成同构表格, 数据只含**系统接管之后**的
账本 (2026-08-28 拍板: 不回填历史, 启用日之前的净值与明细继续查人工表格)。

口径 (逐项对模板核过数, 见 docs 拍板记录):
  * 每行: 交易金额 = 成本价 × 数量; 涨跌幅 = 现价/成本 − 1; 净值估算 = 1 + 收益率。
  * 持仓+平仓收益合计 = 持仓浮动盈亏 Σ + 平仓已实现盈亏 Σ;
    累计净值 = 1 + 该合计 / 净值规模 (参数 PMS_PUBLISH_NAV_SCALE, 0=用 PMS_TOTAL_SCALE)。
  * 可开仓总金额 = 净值规模 + 平仓已实现盈亏 Σ (亏损使其变小, 模板 F107 同法);
    剩余可开仓 = 可开仓总金额 − 存量成本合计; 持仓仓位 = 存量成本合计 / 可开仓总金额。
  * 净值序列每交易日一行, 由调度 15:20 (pms.nav_snapshot) 落 pms_nav_daily;
    导出时当日行用实时价现算覆盖, 保证盘中导出也有今天。
  * 现价取不到的票按成本价顶上并**在表尾如实标注只数** —— 拿不到不装有。

明细行来自批次账 (pms_lot): 持仓明细 = 未平批次 (剩余数量>0), 一行一笔开单;
平仓明细 = 有平仓量的批次 (含部分平仓), 结算价为该批加权平均平仓价。
账户名 / 结构 / 标题是公示口径固定字段, 全部放参数中心 (PMS_PUBLISH_*), 页面可改。
"""
from __future__ import annotations

import io
import logging
from datetime import date, datetime

from app.repo import downstream_repo, pms_repo
from app.services import market, param_store

logger = logging.getLogger("pms.publish")

_FIN_NONE = "无"     # 融资金额/融资费用列: 本产品无融资, 固定"无" (模板同)


# ---------------------------------------------------------------- 取数与口径
def _nav_scale() -> float:
    """净值规模: 参数为 0 时退回总操作规模。两者都为 0 说明参数没配, 直接报错
    比导出一张全是除零的表诚实。"""
    v = param_store.get_float("PMS_PUBLISH_NAV_SCALE", 0.0) or 0.0
    if v <= 0:
        v = param_store.get_float("PMS_TOTAL_SCALE", 0.0) or 0.0
    if v <= 0:
        raise ValueError("净值规模未配置: PMS_PUBLISH_NAV_SCALE 与 PMS_TOTAL_SCALE 都是 0")
    return float(v)


def _as_date(v):
    if isinstance(v, datetime):
        return v.date()
    if isinstance(v, date):
        return v
    return None


def _display_name(v, code: str) -> str:
    """标的列显示名。downstream_repo.fetch_names 返回 {"name": 简称, "full": 全称}
    (取不到的代码它退回 name=代码本身) —— 公示表用简称, 缺简称用全称, 都没有用代码。
    单元格必须是字符串, 字典直接写会被 openpyxl 拒收 (2026-08-28 实机报错修)。"""
    if isinstance(v, dict):
        v = v.get("name") or v.get("full")
    return str(v) if v else code.split(".")[0]


def compute_snapshot() -> dict:
    """组装公示快照: 持仓明细 / 平仓明细 / 汇总 / 当日净值。全部只读。"""
    today = date.today()
    scale = _nav_scale()

    open_lots = pms_repo.list_open_lot_rows()
    closed_lots = pms_repo.list_closed_lot_rows()
    codes = sorted({r["ts_code"] for r in open_lots} | {r["ts_code"] for r in closed_lots})
    try:
        names = downstream_repo.fetch_names(codes) if codes else {}
    except Exception as e:  # 名字取不到不拦导出, 代码列还在
        logger.warning("公示导出取中文名失败 (用代码顶): %s", e)
        names = {}
    open_codes = sorted({r["ts_code"] for r in open_lots})
    prices = market.get_prices(open_codes) if open_codes else {}

    # 字符串参数走通用 get() (ParamStore 没有 get_str; 类型按 RUNTIME_EXTRA 注册项转换)
    account = str(param_store.get("PMS_PUBLISH_ACCOUNT") or "")
    structure = str(param_store.get("PMS_PUBLISH_STRUCTURE") or "")

    holdings, price_missing = [], set()
    for r in open_lots:
        code = r["ts_code"]
        qty = int(r.get("qty") or 0)
        cost = float(r.get("open_price") or 0)
        px = prices.get(code)
        price_ok = bool(px and px > 0)
        if not price_ok:
            px = cost                      # 顶价只为市值可算; 缺价只数在表尾如实标注
            price_missing.add(code)
        od = _as_date(r.get("open_date")) or today
        chg = (px / cost - 1.0) if cost > 0 else 0.0
        holdings.append({
            "account": account, "code": code.split(".")[0],
            "name": _display_name(names.get(code), code),
            "amount": round(cost * qty, 2), "open_date": od, "upd_date": today,
            "structure": structure, "cost": cost, "qty": qty, "price": round(px, 3),
            "days": (today - od).days, "per_share": round(px - cost, 3),
            "chg": chg, "pnl": round((px - cost) * qty, 2), "ret": chg, "nav_est": 1 + chg,
        })

    closed = []
    for r in closed_lots:
        code = r["ts_code"]
        cq = int(r.get("closed_qty") or 0)
        cost = float(r.get("open_price") or 0)
        settle = float(r.get("close_avg_price") or 0)
        pnl = float(r.get("realized_pnl") or 0)
        od = _as_date(r.get("open_date")) or today
        cd = _as_date(r.get("updated_at")) or today
        amt = cost * cq
        ret = (pnl / amt) if amt > 0 else 0.0
        closed.append({
            "account": account, "code": code.split(".")[0],
            "name": _display_name(names.get(code), code),
            "amount": round(amt, 2), "open_date": od, "close_date": cd,
            "structure": structure, "cost": cost, "qty": cq, "settle": round(settle, 3),
            "days": (cd - od).days, "per_share": round(settle - cost, 3),
            "chg": (settle / cost - 1.0) if cost > 0 else 0.0,
            "pnl": round(pnl, 2), "ret": ret, "nav_est": 1 + ret,
        })

    hold_cost = sum(h["amount"] for h in holdings)
    hold_pnl = sum(h["pnl"] for h in holdings)
    realized = sum(c["pnl"] for c in closed)
    total_pnl = hold_pnl + realized
    nav = 1.0 + total_pnl / scale
    openable = scale + realized            # 可开仓总金额 = 净值规模 + 已实现盈亏 (模板口径)
    return {
        "today": today, "scale": scale, "holdings": holdings, "closed": closed,
        "hold_cost": round(hold_cost, 2), "hold_pnl": round(hold_pnl, 2),
        "closed_cost": round(sum(c["amount"] for c in closed), 2),
        "realized": round(realized, 2), "total_pnl": round(total_pnl, 2),
        "nav": round(nav, 4), "openable": round(openable, 2),
        "open_room": round(openable - hold_cost, 2),
        "pos_ratio": round(hold_cost / openable, 4) if openable > 0 else None,
        "price_missing": sorted(price_missing),
        "title": str(param_store.get("PMS_PUBLISH_TITLE") or "量化产品基本信息"),
    }


def record_nav_snapshot() -> dict:
    """净值快照落库 (调度 15:20 调用; 同日重跑覆盖)。失败抛给调度守卫记 ERROR。"""
    s = compute_snapshot()
    ymd = int(s["today"].strftime("%Y%m%d"))
    pms_repo.upsert_nav_daily(
        ymd=ymd, nav=s["nav"], pos_ratio=s["pos_ratio"], holding_pnl=s["hold_pnl"],
        realized_pnl=s["realized"], nav_scale=s["scale"],
        price_missing=len(s["price_missing"]))
    return {"ok": True, "ymd": ymd, "nav": s["nav"], "pos_ratio": s["pos_ratio"],
            "price_missing": len(s["price_missing"])}


def _nav_series(snapshot: dict) -> list:
    """净值序列 = 库里逐日快照 + 当日实时行 (同日覆盖, 盘中导出也有今天)。"""
    ymd_today = int(snapshot["today"].strftime("%Y%m%d"))
    rows = []
    try:
        rows = pms_repo.list_nav_daily()
    except Exception as e:  # 表还没建好 (未跑 init_db) 时导出仍可用, 只有当日一行
        logger.warning("读净值序列失败 (导出只含当日): %s", e)
    out = [r for r in rows if int(r["ymd"]) != ymd_today]
    out.append({"ymd": ymd_today, "nav": snapshot["nav"],
                "pos_ratio": snapshot["pos_ratio"]})
    return out


# ---------------------------------------------------------------- 版式
def build_workbook(s: dict, nav_rows: list) -> bytes:
    """按公司模板版式生成工作簿。纯函数 (不碰库), 可离线测试。"""
    from openpyxl import Workbook
    from openpyxl.chart import LineChart, Reference
    from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
    from openpyxl.utils import get_column_letter

    wb = Workbook()
    ws = wb.active
    ws.title = "Sheet1"
    font = Font(name="等线", size=11)
    bold = Font(name="等线", size=11, bold=True)
    center = Alignment(horizontal="center", vertical="center", wrap_text=True)
    thin = Side(style="thin", color="999999")
    box = Border(left=thin, right=thin, top=thin, bottom=thin)
    head_fill = PatternFill("solid", fgColor="DDEBF7")

    D_FMT, M_FMT, P_FMT, N_FMT = "yyyy/m/d", "#,##0.00", "0.00%", "0.0000"
    HEAD_H = ["序号", "开单账户", "代码", "标的", "交易金额（元）", "起始时间", "更新时间",
              "结构", "交易成本价", "持股数量", "当日价格", "自然天数", "融资金额",
              "融资费用", "每股较期初价盈（元）", "较期初价涨跌幅", "总持股\n浮动盈亏（元）",
              "持有收益率", "净值估算"]
    HEAD_C = HEAD_H.copy()
    HEAD_C[6], HEAD_C[10] = "平仓时间", "结算价"
    FMTS = [None, None, "@", None, M_FMT, D_FMT, D_FMT, None, "0.000", "#,##0", "0.000",
            "0", None, None, "0.000", P_FMT, M_FMT, P_FMT, N_FMT]

    def put(row, col, value, *, f=font, fmt=None, align=None, fill=None, border=box):
        c = ws.cell(row=row, column=col, value=value)
        c.font = f
        if fmt:
            c.number_format = fmt
        if align:
            c.alignment = align
        if fill:
            c.fill = fill
        if border:
            c.border = border
        return c

    def header_row(row, heads):
        for i, h in enumerate(heads):
            put(row, 3 + i, h, f=bold, align=center, fill=head_fill)

    def entry_row(row, e, *, date2_key, px_key):
        vals = [e.get("seq"), e["account"], e["code"], e["name"], e["amount"],
                e["open_date"], e[date2_key], e["structure"], e["cost"], e["qty"],
                e[px_key], e["days"], _FIN_NONE, _FIN_NONE, e["per_share"], e["chg"],
                e["pnl"], e["ret"], e["nav_est"]]
        for i, v in enumerate(vals):
            put(row, 3 + i, v, fmt=FMTS[i])

    def subtotal_row(row, top, bottom):
        """小计行: 金额与盈亏用 SUM 公式, 比率按公式引用 (空表保护为 0)。"""
        put(row, 5, "小计", f=bold, align=center)
        put(row, 7, f"=SUM(G{top}:G{bottom})" if bottom >= top else 0, f=bold, fmt=M_FMT)
        put(row, 18, "合计", f=bold, align=center)
        put(row, 19, f"=SUM(S{top}:S{bottom})" if bottom >= top else 0, f=bold, fmt=M_FMT)
        put(row, 20, f"=IF(G{row}=0,0,S{row}/G{row})", f=bold, fmt=P_FMT)
        put(row, 21, f"=1+T{row}", f=bold, fmt=N_FMT)

    # ---- 标题与时间 ----
    ws.merge_cells(start_row=1, start_column=2, end_row=1, end_column=21)
    put(1, 2, s["title"], f=Font(name="等线", size=16, bold=True), align=center, border=None)
    start = nav_rows[0]["ymd"] if nav_rows else int(s["today"].strftime("%Y%m%d"))
    put(2, 19, "起始时间：", border=None)
    put(2, 20, datetime.strptime(str(start), "%Y%m%d").date(), fmt=D_FMT, border=None)
    put(3, 19, "更新时间：", border=None)
    put(3, 20, s["today"], fmt=D_FMT, border=None)

    # ---- 持仓明细 ----
    r = 4
    header_row(r, HEAD_H)
    hold_top = r + 1
    for i, e in enumerate(s["holdings"], 1):
        e["seq"] = i
        entry_row(r + i, e, date2_key="upd_date", px_key="price")
    if not s["holdings"]:
        put(hold_top, 3, "（当前无持仓）", align=center)
    r_sub = hold_top + (len(s["holdings"]) if s["holdings"] else 1)
    subtotal_row(r_sub, hold_top, r_sub - 1 if s["holdings"] else hold_top - 1)
    ws.merge_cells(start_row=4, start_column=2, end_row=r_sub, end_column=2)
    put(4, 2, "平层\n持仓", f=bold, align=center)

    # ---- 平仓明细 ----
    r = r_sub + 1
    header_row(r, HEAD_C)
    close_top = r + 1
    for i, e in enumerate(s["closed"], 1):
        e["seq"] = i
        entry_row(r + i, e, date2_key="close_date", px_key="settle")
    if not s["closed"]:
        put(close_top, 3, "（暂无平仓记录）", align=center)
    r_csub = close_top + (len(s["closed"]) if s["closed"] else 1)
    subtotal_row(r_csub, close_top, r_csub - 1 if s["closed"] else close_top - 1)
    ws.merge_cells(start_row=r_sub + 1, start_column=2, end_row=r_csub, end_column=2)
    put(r_sub + 1, 2, "平层\n卖出", f=bold, align=center)

    # ---- 汇总两行 (公式引用两张小计行, 口径同模板) ----
    r1, r2 = r_csub + 1, r_csub + 2
    put(r1, 2, "存量合计", f=bold, align=center)
    put(r1, 7, f"=G{r_sub}", f=bold, fmt=M_FMT)
    ws.merge_cells(start_row=r1, start_column=17, end_row=r1, end_column=18)
    put(r1, 17, "持仓+平仓收益合计", f=bold, align=center)
    put(r1, 19, f"=S{r_sub}+S{r_csub}", f=bold, fmt=M_FMT)
    put(r1, 20, f"=S{r1}/{s['scale']}", f=bold, fmt=P_FMT)
    put(r1, 21, f"=1+T{r1}", f=bold, fmt=N_FMT)
    ws.merge_cells(start_row=r2, start_column=2, end_row=r2, end_column=5)
    put(r2, 2, "初始规模+已回收益后可开仓总金额", f=bold, align=center)
    put(r2, 6, f"={s['scale']}+S{r_csub}", f=bold, fmt=M_FMT)
    put(r2, 8, "剩余可开仓金额", f=bold, align=center)
    put(r2, 10, f"=F{r2}-G{r1}", f=bold, fmt=M_FMT)
    put(r2, 11, "持仓仓位", f=bold, align=center)
    put(r2, 12, f"=IF(F{r2}=0,0,G{r1}/F{r2})", f=bold, fmt=P_FMT)
    put(r2, 13, "累计净值", f=bold, align=center)
    put(r2, 15, f"=U{r1}", f=bold, fmt=N_FMT)
    if s["price_missing"]:
        put(r2 + 1, 2, f"注：{len(s['price_missing'])} 只标的当日无行情，按成本价计入市值"
                       f"（{'、'.join(s['price_missing'][:5])}"
                       f"{' 等' if len(s['price_missing']) > 5 else ''}）。",
            border=None)

    # ---- 净值序列 (表右侧, 位置同模板 Y/Z/AA 列) + 折线图 ----
    NC = 25                                # Y 列
    put(4, NC, "时间", f=bold, align=center, fill=head_fill)
    put(4, NC + 1, "净值数据", f=bold, align=center, fill=head_fill)
    put(4, NC + 2, "仓位占比", f=bold, align=center, fill=head_fill)
    for i, row in enumerate(nav_rows, 1):
        put(4 + i, NC, datetime.strptime(str(row["ymd"]), "%Y%m%d").date(), fmt=D_FMT)
        put(4 + i, NC + 1, float(row["nav"]), fmt=N_FMT)
        pr = row.get("pos_ratio")
        put(4 + i, NC + 2, float(pr) if pr is not None else None, fmt=P_FMT)
    if nav_rows:
        chart = LineChart()
        chart.title = "净值走势"
        chart.height, chart.width = 8, 16
        chart.y_axis.title = "净值"
        data = Reference(ws, min_col=NC + 1, min_row=4, max_row=4 + len(nav_rows))
        cats = Reference(ws, min_col=NC, min_row=5, max_row=4 + len(nav_rows))
        chart.add_data(data, titles_from_data=True)
        chart.set_categories(cats)
        ws.add_chart(chart, f"{get_column_letter(NC)}{6 + len(nav_rows)}")

    # ---- 列宽 ----
    widths = {2: 9, 3: 5, 4: 10, 5: 9, 6: 13, 7: 11, 8: 11, 9: 12, 10: 9, 11: 9, 12: 9,
              13: 9, 14: 9, 15: 10, 16: 12, 17: 12, 18: 13, 19: 11, 20: 11, 21: 9,
              25: 11, 26: 10, 27: 10}
    for col, w in widths.items():
        ws.column_dimensions[get_column_letter(col)].width = w
    ws.freeze_panes = "C5"

    buf = io.BytesIO()
    wb.save(buf)
    return buf.getvalue()


def export_xlsx() -> tuple:
    """导出入口: 返回 (文件名, xlsx 字节)。文件名沿用公司习惯: 量化数据YYYY.M.D.xlsx。"""
    s = compute_snapshot()
    blob = build_workbook(s, _nav_series(s))
    t = s["today"]
    return f"量化数据{t.year}.{t.month}.{t.day}.xlsx", blob