330 lines
13 KiB
Python
330 lines
13 KiB
Python
"""交割单解析（券商导出的成交流水 → 结构化买卖记录）。
|
||
|
||
支持三类导出物（按内容嗅探，不信任扩展名）：
|
||
- CSV/制表符文本（utf-8-sig / gbk / gb18030 自动探测）
|
||
- Excel .xlsx（openpyxl；很多券商导出的 .xls 实为 xlsx 或 HTML，先按魔数分流）
|
||
- HTML 表格（.xls 常见真身：<table><tr><td>）
|
||
|
||
列名模糊匹配兼容通达信/恒生/同花顺系的命名差异；业务名称含「买入/卖出」
|
||
才入库，银行转账、配号、利息、红利等非交易行跳过并计数。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import csv
|
||
import io
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
|
||
from fastapi import HTTPException
|
||
|
||
|
||
@dataclass
|
||
class ParsedTrade:
|
||
trade_date: date
|
||
ts_code: str
|
||
code: str
|
||
name: str
|
||
direction: str # buy | sell
|
||
price: float | None
|
||
qty: float
|
||
amount: float | None
|
||
fee: float
|
||
raw: dict = field(default_factory=dict)
|
||
|
||
|
||
@dataclass
|
||
class ParseResult:
|
||
trades: list[ParsedTrade] = field(default_factory=list)
|
||
skipped_other: int = 0 # 非证券买卖行（转账/配号/利息等）
|
||
skipped_bad: list[str] = field(default_factory=list) # 解析失败样例（截断到前 5 条）
|
||
header_row_index: int = -1
|
||
columns: dict[str, str] = field(default_factory=dict) # 逻辑列 -> 实际列名
|
||
|
||
|
||
# ---------- 列名别名（归一化后做「包含」匹配，先命中的优先） ----------
|
||
COLUMN_ALIASES: dict[str, list[str]] = {
|
||
"date": ["成交日期", "交割日期", "交收日期", "交易日期", "过户日期", "发生日期", "清算日期", "日期"],
|
||
"op": ["业务名称", "业务摘要", "操作", "业务类型", "交易类型", "交易类别", "摘要", "方向", "买卖标志"],
|
||
"code": ["证券代码", "股票代码", "产品代码", "代码"],
|
||
"name": ["证券名称", "股票名称", "产品名称", "名称"],
|
||
"qty": ["成交数量", "发生数量", "委托数量", "成交股数", "数量"],
|
||
"price": ["成交价格", "成交均价", "成交价", "均价", "价格"],
|
||
"amount": ["成交金额", "成交清算金额", "清算金额", "发生金额", "资金发生数", "金额"],
|
||
"fee": ["手续费", "佣金", "印花税", "过户费", "其他费", "杂费", "规费"],
|
||
}
|
||
# 手续费类允许多列求和（手续费+印花税+过户费…），其余逻辑列取第一命中
|
||
_FEE_KEYS = ("手续费", "佣金", "印花税", "过户费", "其他费", "杂费", "规费")
|
||
|
||
|
||
def _norm_header(h: str) -> str:
|
||
"""列名归一化：去空白、去全角、去括号单位（如「成交数量(股)」）。"""
|
||
h = str(h).strip().replace(" ", "").replace(" ", "").replace(" ", "")
|
||
h = re.sub(r"[（(【\[].*?[)）】\]]", "", h)
|
||
return h
|
||
|
||
|
||
def _match_columns(header: list[str]) -> dict[str, str]:
|
||
"""表头 -> 逻辑列映射。返回 {逻辑列: 实际列名}；费率类列全部收集到 fee(合并名)。"""
|
||
out: dict[str, str] = {}
|
||
fee_cols: list[str] = []
|
||
for h in header:
|
||
n = _norm_header(h)
|
||
if not n:
|
||
continue
|
||
for key, aliases in COLUMN_ALIASES.items():
|
||
if key == "fee":
|
||
if any(a in n for a in _FEE_KEYS):
|
||
fee_cols.append(h)
|
||
continue
|
||
if key in out:
|
||
continue
|
||
if any(a in n for a in aliases):
|
||
out[key] = h
|
||
break
|
||
# 「费用合计」列本身已含全部费用明细，取它即可，避免与手续费/印花税等列重复累加
|
||
total_col = next((h for h in header if "费用合计" in _norm_header(h)), None)
|
||
if total_col is not None:
|
||
out["fee"] = total_col
|
||
elif fee_cols:
|
||
out["fee"] = "\x00".join(fee_cols) # 多列合并存储，取值时拆开求和
|
||
return out
|
||
|
||
|
||
def _looks_like_header(row: list[str]) -> bool:
|
||
"""前 10 行里找表头：≥3 个逻辑列可识别即认为是表头。"""
|
||
return len(_match_columns(row)) >= 3
|
||
|
||
|
||
def _to_float(v) -> float | None:
|
||
"""'1,234.50' / '(123.45)' / '--' / '' → float；不可解析返回 None。"""
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, (int, float)):
|
||
return float(v)
|
||
s = str(v).strip().replace(",", "").replace("，", "")
|
||
if not s or s in {"--", "-", "—"}:
|
||
return None
|
||
neg = s.startswith("(") and s.endswith(")")
|
||
if neg:
|
||
s = s[1:-1]
|
||
try:
|
||
f = float(s)
|
||
except ValueError:
|
||
return None
|
||
return -f if neg else f
|
||
|
||
|
||
def _to_date(v) -> date | None:
|
||
if isinstance(v, datetime):
|
||
return v.date()
|
||
if isinstance(v, date):
|
||
return v
|
||
if isinstance(v, (int, float)) and not isinstance(v, bool) and 30000 < v < 60000:
|
||
# Excel 日期序列值（1982~2064），openpyxl 读无日期格式的单元格时会给出
|
||
from datetime import timedelta
|
||
return date(1899, 12, 30) + timedelta(days=int(v))
|
||
s = str(v).strip()
|
||
m = re.search(r"(\d{4})[-/.年](\d{1,2})[-/.月](\d{1,2})", s)
|
||
if not m:
|
||
m2 = re.fullmatch(r"(\d{4})(\d{2})(\d{2})", s)
|
||
if not m2:
|
||
return None
|
||
m = m2
|
||
y, mo, d = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
||
try:
|
||
return date(y, mo, d)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _to_code_suffix(code: str) -> str:
|
||
"""6 位代码 → 交易所后缀（60/68 沪，00/30 深，4/8/92 北交所）。"""
|
||
if code.startswith(("60", "68", "90")):
|
||
return ".SH"
|
||
if code.startswith(("00", "30", "20")):
|
||
return ".SZ"
|
||
return ".BJ"
|
||
|
||
|
||
def _direction(op: str) -> str | None:
|
||
s = str(op)
|
||
if "买入" in s or "buy" in s.lower() or "证券买" in s:
|
||
return "buy"
|
||
if "卖出" in s or "sell" in s.lower() or "证券卖" in s:
|
||
return "sell"
|
||
return None
|
||
|
||
|
||
def _parse_rows(rows: list[list[object]]) -> ParseResult:
|
||
"""已抽成二维表的行集 → ParseResult。rows[0] 应是表头（调用方已定位）。"""
|
||
res = ParseResult()
|
||
if not rows:
|
||
return res
|
||
header = [str(h) for h in rows[0]]
|
||
cols = _match_columns(header)
|
||
res.columns = {k: v for k, v in cols.items()}
|
||
res.header_row_index = 0
|
||
need = ("date", "qty")
|
||
if not all(k in cols for k in need) or not ("code" in cols or "name" in cols):
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail="识别不到交割单表头（需要 成交日期/证券代码或证券名称/成交数量 等列），"
|
||
"请确认导出的是「交割单/历史成交」文件",
|
||
)
|
||
idx = {h: i for i, h in enumerate(header)}
|
||
|
||
# 无「业务名称」列的导出（如部分招商证券格式）：靠发生金额正负判方向（买入为负）。
|
||
# 仅当数据里确实存在负数金额才启用，避免「全正数」格式被误判。
|
||
def _amount_of(row: list[object]) -> float | None:
|
||
i = idx.get(cols["amount"])
|
||
return _to_float(row[i]) if i is not None and i < len(row) else None
|
||
|
||
sign_mode = "op" not in cols and "amount" in cols and any(
|
||
(_amount_of(row) or 0) < 0 for row in rows[1:] if any(str(c).strip() for c in row)
|
||
)
|
||
|
||
def cell(row: list[object], col: str):
|
||
i = idx.get(col)
|
||
return row[i] if i is not None and i < len(row) else None
|
||
|
||
for row in rows[1:]:
|
||
d = _to_date(cell(row, cols["date"]))
|
||
code = re.sub(r"\D", "", str(cell(row, cols["code"]) or "")) if "code" in cols else ""
|
||
raw_amount = _amount_of(row) if sign_mode else None
|
||
direction = (
|
||
_direction(str(cell(row, cols["op"]) or "")) if "op" in cols
|
||
else ("buy" if (raw_amount or 0) < 0 else "sell") if sign_mode
|
||
else None
|
||
)
|
||
name = str(cell(row, cols["name"]) or "").strip() if "name" in cols else ""
|
||
if d is None or (not code and not name) or direction is None:
|
||
# 无日期/无代码且无名称/非买卖业务（银行转账、配号、利息、红利等）
|
||
if any(str(c).strip() for c in row):
|
||
res.skipped_other += 1
|
||
continue
|
||
if len(code) > 6:
|
||
code = code[-6:] # 个别导出带市场前缀（如 1:600000 / sh600000）
|
||
qty = abs(_to_float(cell(row, cols["qty"])) or 0)
|
||
if qty <= 0:
|
||
res.skipped_bad.append(f"{d} {code or name} 数量无效：{cell(row, cols['qty'])!r}")
|
||
continue
|
||
price = _to_float(cell(row, cols["price"])) if "price" in cols else None
|
||
amount = raw_amount if sign_mode else (_to_float(cell(row, cols["amount"])) if "amount" in cols else None)
|
||
if amount is not None:
|
||
amount = abs(amount)
|
||
fee = 0.0
|
||
if "fee" in cols:
|
||
for fc in cols["fee"].split("\x00"):
|
||
f = _to_float(cell(row, fc))
|
||
if f:
|
||
fee += abs(f)
|
||
# 无代码列（招商式导出）：ts_code 留空，由 API 层按 name 反查 stock_basic
|
||
ts_code = code + _to_code_suffix(code) if code else ""
|
||
res.trades.append(ParsedTrade(
|
||
trade_date=d,
|
||
code=code,
|
||
ts_code=ts_code,
|
||
name=name,
|
||
direction=direction,
|
||
price=price,
|
||
qty=qty,
|
||
amount=amount,
|
||
fee=round(fee, 2),
|
||
raw={h: row[i] if i < len(row) else None for i, h in enumerate(header)},
|
||
))
|
||
res.skipped_bad = res.skipped_bad[:5]
|
||
return res
|
||
|
||
|
||
def _find_header(rows: list[list[object]]) -> int:
|
||
for i, row in enumerate(rows[:10]):
|
||
if _looks_like_header([str(c) for c in row]):
|
||
return i
|
||
return -1
|
||
|
||
|
||
# ---------- 输入格式分流 ----------
|
||
def _rows_from_csv(data: bytes) -> list[list[object]]:
|
||
"""逗号/制表符分隔文本。sniff 分隔符；跳过全空行。"""
|
||
text = None
|
||
for enc in ("utf-8-sig", "gbk", "gb18030"):
|
||
try:
|
||
text = data.decode(enc)
|
||
break
|
||
except UnicodeDecodeError:
|
||
continue
|
||
if text is None:
|
||
raise HTTPException(status_code=422, detail="文件编码无法识别（支持 UTF-8 / GBK）")
|
||
sample = text[:4096]
|
||
delim = "\t" if sample.count("\t") > sample.count(",") else ","
|
||
lines = [ln for ln in text.splitlines() if ln.strip()]
|
||
if not lines:
|
||
raise HTTPException(status_code=422, detail="文件是空的")
|
||
return [next(csv.reader([ln], delimiter=delim)) for ln in lines]
|
||
|
||
|
||
def _rows_from_xlsx(data: bytes) -> list[list[object]]:
|
||
from openpyxl import load_workbook
|
||
|
||
try:
|
||
wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
|
||
except Exception as e: # noqa: BLE001 - openpyxl 对损坏文件抛各种类型
|
||
raise HTTPException(status_code=422, detail=f"Excel 文件无法读取：{e}") from e
|
||
ws = wb.active
|
||
rows = [[c for c in row] for row in ws.iter_rows(values_only=True)]
|
||
wb.close()
|
||
return rows
|
||
|
||
|
||
_TD_RE = re.compile(r"<t[dh][^>]*>(.*?)</t[dh]>", re.IGNORECASE | re.DOTALL)
|
||
_TR_RE = re.compile(r"<tr[^>]*>(.*?)</tr>", re.IGNORECASE | re.DOTALL)
|
||
|
||
|
||
def _rows_from_html(data: bytes) -> list[list[object]]:
|
||
"""券商导出的 .xls 常是 HTML 表格。去掉标签实体后按 <tr>/<td> 切。"""
|
||
text = None
|
||
for enc in ("utf-8", "gbk", "gb18030"):
|
||
try:
|
||
text = data.decode(enc)
|
||
break
|
||
except UnicodeDecodeError:
|
||
continue
|
||
if text is None:
|
||
raise HTTPException(status_code=422, detail="文件编码无法识别（支持 UTF-8 / GBK）")
|
||
import html as html_mod
|
||
|
||
rows: list[list[object]] = []
|
||
for tr in _TR_RE.findall(text):
|
||
cells = [html_mod.unescape(re.sub(r"<[^>]+>", "", td)).strip() for td in _TD_RE.findall(tr)]
|
||
rows.append(cells)
|
||
if not rows:
|
||
raise HTTPException(status_code=422, detail="HTML 里没有表格数据")
|
||
return rows
|
||
|
||
|
||
def parse_statement(data: bytes, filename: str) -> ParseResult:
|
||
"""入口：按内容魔数/特征分流 → 定位表头 → 解析。"""
|
||
if not data:
|
||
raise HTTPException(status_code=422, detail="文件是空的")
|
||
head = data[:512].lstrip()
|
||
if head.startswith(b"PK"):
|
||
rows = _rows_from_xlsx(data)
|
||
elif head[:1] in (b"<",) or head.lower().startswith(b"\xef\xbb\xbf<"):
|
||
rows = _rows_from_html(data)
|
||
elif filename.lower().endswith((".xlsx", ".xls")) and not head.startswith((b"PK", b"<")):
|
||
# 扩展名是 Excel 但内容既非 xlsx 也非 HTML → 试试当文本
|
||
rows = _rows_from_csv(data)
|
||
else:
|
||
rows = _rows_from_csv(data)
|
||
# 去尾部全空行，定位表头（导出物常有标题行/账户信息行在前）
|
||
while rows and not any(str(c).strip() for c in rows[-1]):
|
||
rows.pop()
|
||
hi = _find_header(rows)
|
||
if hi < 0:
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail="找不到表头行（前 10 行内没有 成交日期/证券代码 等列名），请确认导出的是交割单",
|
||
)
|
||
return _parse_rows(rows[hi:])
|