"""交割单解析器离线自测：不碰数据库，直接调 app.trades.parse_statement。 覆盖四类真实导出格式 + 边界行（转账/配号/利息跳过、费用合计列去重、日期多格式）。 运行：uv run python scripts/test_trades_parser.py """ from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from app.trades import parse_statement # noqa: E402 FAIL: list[str] = [] def check(name: str, cond: bool, detail: str = "") -> None: mark = "ok " if cond else "FAIL" print(f"[{mark}] {name}{(' — ' + detail) if detail and not cond else ''}") if not cond: FAIL.append(name) # ---------- 1) 通达信式：GBK + 制表符 + 标题行在前 ---------- tdx = ( "交割单\n" "股东账号: A123456789 起始日期: 20240102 终止日期: 20240105 币种: 人民币\n" "\t交割日期\t业务名称\t证券代码\t证券名称\t成交价格\t成交数量\t成交金额\t手续费\t印花税\t过户费\t发生金额\t资金余额\t合同号\n" "\t20240102\t证券买入\t600519\t贵州茅台\t1680.00\t100\t168000.00\t5.00\t0.00\t1.68\t-168006.68\t200000.00\t1000001\n" "\t20240102\t银行转存\t\t\t\t\t\t\t\t\t50000.00\t250000.00\t\n" "\t20240103\t证券卖出\t600519\t贵州茅台\t1700.50\t100\t170050.00\t5.00\t170.05\t1.70\t169873.25\t419873.25\t1000002\n" "\t20240105\t利息归本\t\t\t\t\t\t\t\t\t1.25\t419874.50\t\n" ) r = parse_statement(tdx.encode("gbk"), "交割单.txt") check("tdx: 2 笔成交", len(r.trades) == 2, f"got {len(r.trades)}") check("tdx: 跳过 2 行非交易", r.skipped_other == 2, f"got {r.skipped_other}") t0, t1 = r.trades[0], r.trades[1] check("tdx: 日期/代码/后缀", (t0.trade_date.isoformat(), t0.ts_code) == ("2024-01-02", "600519.SH"), f"{t0.trade_date} {t0.ts_code}") check("tdx: 买入方向+费用合计", t0.direction == "buy" and abs(t0.fee - 6.68) < 1e-9, f"{t0.direction} fee={t0.fee}") check("tdx: 卖出费用含印花税", t1.direction == "sell" and abs(t1.fee - 176.75) < 1e-9, f"fee={t1.fee}") check("tdx: 金额取绝对值", t0.amount == 168000.0, f"amount={t0.amount}") # ---------- 2) 恒生柜台式：UTF-8 CSV，交收日期/交易类别/费用合计 ---------- hs = ( "序号,交收日期,证券代码,证券名称,交易类别,成交价格,成交数量,证券余额,成交金额,资金发生数,资金余额,流水序号,业务标志,业务名称,发生金额,后资金额,货币类别,费用合计,净佣金,规费,印花税,过户费,合同号\n" "1,2024-06-07,000858,五粮液,证券买入,132.50,200,200,26500.00,-26505.80,73494.20,1,0101,证券买入,-26505.80,73494.20,人民币,5.80,4.20,1.60,0.00,0.00,66778001\n" "2,2024-06-07,,,\t,,,,5120.00,78614.20,2,2041,银行转存,5120.00,78614.20,人民币,0,0,0,0,0,\n" "3,2024-06-10,000858,五粮液,证券卖出,135.00,200,0,27000.00,26975.30,105589.50,3,0102,证券卖出,26975.30,105589.50,人民币,24.70,4.20,1.60,18.90,0.00,66779001\n" ) r2 = parse_statement(hs.encode("utf-8"), "hsi.csv") check("hs: 2 笔成交", len(r2.trades) == 2, f"got {len(r2.trades)}") check("hs: 费用合计不重复累加", abs(r2.trades[1].fee - 24.70) < 1e-9, f"fee={r2.trades[1].fee}") check("hs: 深市后缀", r2.trades[0].ts_code == "000858.SZ", r2.trades[0].ts_code) check("hs: 日期 YYYY-MM-DD", r2.trades[0].trade_date.isoformat() == "2024-06-07") # ---------- 3) HTML 伪 .xls（同花顺导出常见真身） ---------- html = """
客户姓名测试
成交日期业务名称证券代码证券名称成交价格成交数量成交金额手续费
2024/03/15证券买入300750宁德时代182.3030054,690.0016.41
2024/03/18证券卖出300750宁德时代185.0030055,500.005.55
""" r3 = parse_statement(html.encode("gbk"), "jiaogedan.xls") check("html: 2 笔成交", len(r3.trades) == 2, f"got {len(r3.trades)}") check("html: 千分位金额", r3.trades[0].amount == 54690.0, f"{r3.trades[0].amount}") check("html: 创业板后缀", r3.trades[0].ts_code == "300750.SZ", r3.trades[0].ts_code) check("html: 斜杠日期", r3.trades[1].trade_date.isoformat() == "2024-03-18") # ---------- 4) 无业务名称列：发生金额正负判方向（招商式） ---------- zh = ( "证券名称,成交日期,成交价格,成交数量,发生金额,资金余额,合同编号\n" "贵州茅台,20240102,1680.00,100,-168005.00,200000.00,SZ1000001\n" "贵州茅台,20240103,1700.50,100,170049.50,370049.50,SZ1000002\n" ) r4 = parse_statement(zh.encode("utf-8"), "zszs.csv") check("sign: 2 笔成交", len(r4.trades) == 2, f"got {len(r4.trades)}") check("sign: 负金额=买入", (r4.trades[0].direction, r4.trades[1].direction) == ("buy", "sell"), f"{r4.trades[0].direction}/{r4.trades[1].direction}") # ---------- 5) xlsx（openpyxl 内存构造） ---------- import io # noqa: E402 from openpyxl import Workbook # noqa: E402 wb = Workbook() ws = wb.active ws.append(["对账单", None, None]) ws.append(["成交日期", "业务名称", "证券代码", "证券名称", "成交均价", "成交股数", "成交金额", "佣金", "过户费"]) from datetime import datetime as dt # noqa: E402 ws.append([dt(2024, 2, 28, 14, 35, 0), "证券买入", "688981", "中芯国际", 52.80, 200, 10560.00, 2.50, 1.06]) ws.append([dt(2024, 3, 1, 9, 31, 0), "证券卖出", "688981", "中芯国际", 54.10, 200, 10820.00, 2.50, 1.06]) buf = io.BytesIO() wb.save(buf) r5 = parse_statement(buf.getvalue(), "sm.xlsx") check("xlsx: 2 笔成交", len(r5.trades) == 2, f"got {len(r5.trades)}") check("xlsx: datetime 日期", r5.trades[0].trade_date.isoformat() == "2024-02-28") check("xlsx: 科创板后缀", r5.trades[0].ts_code == "688981.SH", r5.trades[0].ts_code) check("xlsx: 佣金+过户费", abs(r5.trades[0].fee - 3.56) < 1e-9, f"fee={r5.trades[0].fee}") # ---------- 6) 错误分支 ---------- from fastapi import HTTPException # noqa: E402 try: parse_statement("随便一串不是交割单的文字,1,2,3".encode("utf-8"), "x.csv") check("garbage: 应 422", False) except HTTPException as e: check("garbage: 422", e.status_code == 422) print() if FAIL: print(f"FAIL {len(FAIL)}: {FAIL}") sys.exit(1) print("PASS: 交割单解析器全部用例通过")