dekun
92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
# Copyright (c) 2025-2026 马建军. All rights reserved.
|
|
# 详见 LICENSE.zh-CN.txt
|
|
|
|
"""CTP 持仓均价：成交加权 / 柜台持仓价（滚仓加仓后以柜台为准）。"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from ctp_symbol import ths_to_vnpy_symbol
|
|
|
|
|
|
def symbols_match(ctp_sym: str, ths: str) -> bool:
|
|
a = (ctp_sym or "").lower()
|
|
b = (ths or "").lower()
|
|
if a == b:
|
|
return True
|
|
if a and b and a.split(".")[0] == b.split(".")[0]:
|
|
return True
|
|
try:
|
|
vnpy_sym, _ = ths_to_vnpy_symbol(ths)
|
|
if a == vnpy_sym.lower():
|
|
return True
|
|
except Exception:
|
|
pass
|
|
try:
|
|
vnpy_sym, _ = ths_to_vnpy_symbol(ctp_sym)
|
|
if vnpy_sym.lower() == b.split(".")[0]:
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
|
|
def avg_from_trades(
|
|
trades: list[dict[str, Any]],
|
|
sym: str,
|
|
direction: str,
|
|
*,
|
|
expect_lots: int = 0,
|
|
) -> Optional[float]:
|
|
"""按成交回报移动加权均价（滚仓多笔开仓后应与柜台一致）。"""
|
|
direction = (direction or "long").strip().lower()
|
|
vol = 0
|
|
cost = 0.0
|
|
for t in sorted(trades, key=lambda x: (x.get("datetime") or "", x.get("trade_id") or "")):
|
|
if not symbols_match(t.get("symbol") or "", sym):
|
|
continue
|
|
off = (t.get("offset") or "").strip().lower()
|
|
pos_dir = (
|
|
t.get("position_direction") or t.get("direction") or "long"
|
|
).strip().lower()
|
|
if pos_dir != direction:
|
|
continue
|
|
lots = int(t.get("lots") or 0)
|
|
px = float(t.get("price") or 0)
|
|
if lots <= 0 or px <= 0:
|
|
continue
|
|
if off == "open":
|
|
cost += px * lots
|
|
vol += lots
|
|
elif off == "close" and vol > 0:
|
|
avg = cost / vol
|
|
dec = min(lots, vol)
|
|
cost -= avg * dec
|
|
vol -= dec
|
|
if vol <= 0:
|
|
return None
|
|
if expect_lots > 0 and vol != expect_lots:
|
|
return None
|
|
return round(cost / vol, 4)
|
|
|
|
|
|
def resolve_ctp_entry(
|
|
sym: str,
|
|
direction: str,
|
|
ctp: Optional[dict[str, Any]],
|
|
trades: Optional[list[dict[str, Any]]] = None,
|
|
) -> tuple[float, str]:
|
|
"""均价：成交加权 > 柜台 PositionCost 持仓价。"""
|
|
if not ctp:
|
|
return 0.0, "none"
|
|
direction = (direction or "long").strip().lower()
|
|
lots = int(ctp.get("lots") or 0)
|
|
if trades:
|
|
trade_avg = avg_from_trades(trades, sym, direction, expect_lots=lots)
|
|
if trade_avg and trade_avg > 0:
|
|
return float(trade_avg), "trades"
|
|
pos_avg = float(ctp.get("avg_price") or 0)
|
|
if pos_avg > 0:
|
|
return pos_avg, "ctp"
|
|
return 0.0, "none"
|