88 lines
4.3 KiB
Python
88 lines
4.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Perold 落地差距四分量纯函数计算器（spec §4.7 第1/2条）。
|
||
|
||
定性：paper 腿=无摩擦基准（Perold 原教旨=决策时刻以决策价建仓），四分量全部
|
||
定义在 live 腿；paper 每日一次 step 撮合下延迟/冲击≡机制性零——显式标注非缺数。
|
||
分量（正=逆风，方向 buy=+1/sell=−1）：
|
||
延迟=(到达−决策)×方向×已成交量
|
||
冲击=(成交均−到达)×方向×成交量
|
||
机会=(收盘−决策)×方向×未成交量（Perold EOD 口径）
|
||
费用=Σfee（estimate_fee 口径，Task 1）
|
||
可加和恒等式：延迟+冲击+机会+费用 == Σ方向×[(成交均−决策)×qf+(收盘−决策)×qu]+费用
|
||
（identity_resid 透出，>1e-6 即回归）。qf>0 而 fills 空（超时行无逐笔价）时
|
||
恒等式出现缺口=identity_resid 非零——「缺成交明细」的诚实信号，非回归；
|
||
decision/arrival 有而 close 缺（或反之）的腿只进分量侧不进直算侧，resid
|
||
同样非零=缺数信号非回归。
|
||
价格移动参考行（Kissell-Glantz price appreciation，不入 is_total）：
|
||
(收盘−决策)×方向×委托总量——大盘漂移语境，防漂移污染执行评价。
|
||
缺锚（decision/arrival/close 任一 None）：该分量跳过、腿计入 legs_no_anchor
|
||
（显式可见，宁缺毋假），费用照算。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
|
||
PAPER_MECHANISM_ZERO_NOTE = (
|
||
"paper 腿延迟/冲击=机制性零（每日一次 step 撮合，定价=20:30 收盘回放）；"
|
||
"显式标注非缺数，四分量全部定义在 live 腿")
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OrderLeg:
|
||
action: str
|
||
qty: float
|
||
fills: tuple[tuple[float, float], ...] = ()
|
||
filled_qty: float | None = None # 重放权威已成交量（超时行无逐笔价场景）
|
||
decision_price: float | None = None
|
||
arrival_price: float | None = None
|
||
close_price: float | None = None
|
||
reject_reason: str | None = None
|
||
fee: float = 0.0
|
||
|
||
|
||
def compute_components(legs: list[OrderLeg]) -> dict:
|
||
delay = impact = opp = fee_total = movement = direct = 0.0
|
||
anchored = 0
|
||
traded_value = 0.0
|
||
by_reason = {k: 0.0 for k in
|
||
("T1", "CASH", "LIMIT_UP", "LIMIT_DOWN", "TIMEOUT")}
|
||
by_legs = {k: 0 for k in by_reason}
|
||
for leg in legs:
|
||
s = 1.0 if leg.action == "buy" else -1.0
|
||
qf = leg.filled_qty if leg.filled_qty is not None \
|
||
else sum(v for _, v in leg.fills)
|
||
qu = max(leg.qty - qf, 0.0)
|
||
fv = sum(v for _, v in leg.fills)
|
||
fill_avg = (sum(p * v for p, v in leg.fills) / fv) if fv else None
|
||
fee_total += leg.fee
|
||
traded_value += sum(p * v for p, v in leg.fills)
|
||
if leg.reject_reason:
|
||
by_legs[leg.reject_reason] = by_legs.get(leg.reject_reason, 0) + 1
|
||
if leg.decision_price is None:
|
||
continue
|
||
anchored += 1
|
||
dec = leg.decision_price
|
||
if qf and leg.arrival_price is not None:
|
||
delay += s * (leg.arrival_price - dec) * qf
|
||
if fill_avg is not None and leg.arrival_price is not None:
|
||
impact += s * (fill_avg - leg.arrival_price) * qf
|
||
if qu and leg.close_price is not None:
|
||
opp += s * (leg.close_price - dec) * qu
|
||
if leg.reject_reason:
|
||
by_reason[leg.reject_reason] = by_reason.get(
|
||
leg.reject_reason, 0.0) + s * (leg.close_price - dec) * qu
|
||
if leg.close_price is not None:
|
||
movement += s * (leg.close_price - dec) * leg.qty
|
||
if leg.close_price is not None and (fill_avg is not None or qf == 0):
|
||
fill_term = (fill_avg - dec) * qf if fill_avg is not None else 0.0
|
||
direct += s * (fill_term + (leg.close_price - dec) * qu) + leg.fee
|
||
total = delay + impact + opp + fee_total
|
||
return {"delay_cost": delay, "impact_cost": impact,
|
||
"opportunity_cost": opp, "fee_total": fee_total,
|
||
"price_movement": movement, "is_total": total,
|
||
"identity_resid": abs(total - direct),
|
||
"legs_count": len(legs), "legs_no_anchor": len(legs) - anchored,
|
||
"traded_value": traded_value, "by_reason": by_reason,
|
||
"by_reason_legs": by_legs,
|
||
"paper": {"mechanism_zero": True, "note": PAPER_MECHANISM_ZERO_NOTE}}
|