dekun/crypto_okx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

删除永续下单监控与划转强平，保留关键位监控并完成回归审计。

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-10-07 13:04:21 +08:00
parent 7c5a6e25b2
commit c021c38e2d
39 changed files with 5841 additions and 10637 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/trade/compensating_close_lib.py
-16
View File
@@ -1,16 +0,0 @@
"""开仓后挂 TP/SL 失败时的补偿平仓(避免裸仓)."""
from __future__ import annotations
from typing import Callable
def log_compensating_close_error(prefix: str, exc: BaseException) -> None:
print(f"[{prefix}] {exc}", flush=True)
def run_compensating_close(close_fn: Callable[[], None], *, log_prefix: str = "compensating_close") -> None:
"""执行补偿平仓;二次失败只打日志,不掩盖原始异常."""
try:
close_fn()
except Exception as e:
log_compensating_close_error(log_prefix, e)
lib/trade/daily_open_limit_lib.py
-140
View File
@@ -1,140 +0,0 @@
"""单日开仓次数:软提醒阈值 + 硬上限(三所实例共用)."""
from __future__ import annotations
import os
from typing import Any, Optional
def parse_daily_open_alert_threshold(raw: Any = None, *, default: int = 5) -> int:
"""AI 克制提醒阈值;至少 1."""
try:
v = int(raw if raw is not None and str(raw).strip() != "" else default)
except (TypeError, ValueError):
v = default
return max(1, v)
def parse_daily_open_hard_limit(raw: Any = None, *, default: int = 0) -> int:
"""硬上限;0 表示不启用.至少 0."""
try:
v = int(raw if raw is not None and str(raw).strip() != "" else default)
except (TypeError, ValueError):
v = default
return max(0, v)
def load_daily_open_limits_from_env(
env: Optional[dict[str, str]] = None,
) -> tuple[int, int]:
"""从环境变量读取 (alert_threshold, hard_limit)."""
src = env if env is not None else os.environ
alert = parse_daily_open_alert_threshold(src.get("DAILY_OPEN_ALERT_THRESHOLD"))
hard = parse_daily_open_hard_limit(src.get("DAILY_OPEN_HARD_LIMIT"))
return alert, hard
def count_opens_for_trading_day(conn, trading_day: str) -> int:
"""本交易日已成功写入 order_monitors 的开仓次数."""
td = (trading_day or "").strip()
if not td:
return 0
row = conn.execute(
"SELECT COUNT(*) FROM order_monitors WHERE session_date=?",
(td,),
).fetchone()
return int(row[0] if row else 0)
def daily_open_hard_limit_blocks(opens_today: int, hard_limit: int) -> bool:
return int(hard_limit) > 0 and int(opens_today) >= int(hard_limit)
def hard_limit_block_reason(opens_today: int, hard_limit: int, reset_hour: int) -> str:
return (
f"本交易日开仓次数已达上限({int(opens_today)}/{int(hard_limit)}),"
f"次日北京时间 {int(reset_hour)}:00 后恢复"
)
def check_daily_open_hard_limit(
conn,
trading_day: str,
hard_limit: int,
reset_hour: int,
) -> tuple[bool, str, int]:
"""返回 (允许继续开仓, 拒绝原因, 当日已开次数)."""
opens_today = count_opens_for_trading_day(conn, trading_day)
if daily_open_hard_limit_blocks(opens_today, hard_limit):
return False, hard_limit_block_reason(opens_today, hard_limit, reset_hour), opens_today
return True, "", opens_today
def can_trade_new_open(
*,
time_allows: bool,
active_count: int,
max_active_positions: int,
opens_today: int,
hard_limit: int,
extra_blocks: bool = False,
) -> bool:
if extra_blocks:
return False
if not time_allows:
return False
if int(active_count) >= int(max_active_positions):
return False
if daily_open_hard_limit_blocks(opens_today, hard_limit):
return False
return True
def should_send_daily_open_alert(before: int, after: int, alert_threshold: int) -> bool:
return int(before) < int(alert_threshold) <= int(after)
def build_daily_open_alert_prompt(
trading_day: str,
opens_after: int,
alert_threshold: int,
*,
hard_limit: int = 0,
detail_line: str = "",
) -> str:
hard_txt = (
f"硬上限 {hard_limit} 次(已达后将禁止新开仓直至下一交易日)."
if int(hard_limit) > 0
else "未配置单日硬上限."
)
extra = f" {detail_line}" if detail_line else ""
return (
f"用户在北京时间交易日 {trading_day} 已累计开仓 {opens_after} 次"
f"(AI 提醒阈值 {alert_threshold};{hard_txt})"
f"{extra}"
f"用户自述“上头了”.请给克制提醒."
)
def format_daily_open_counter_line(
opens_today: int,
alert_threshold: int,
hard_limit: int,
) -> str:
if int(hard_limit) > 0:
return (
f"📅 当日开仓次数:{int(opens_today)} / 硬上限 {int(hard_limit)} 次"
f"(AI 提醒阈值 {int(alert_threshold)})"
)
return (
f"📅 当日开仓次数:{int(opens_today)} / AI 提醒阈值 {int(alert_threshold)} 次"
)
def format_daily_open_summary_short(
opens_today: int,
alert_threshold: int,
hard_limit: int,
) -> str:
if int(hard_limit) > 0:
return f"本交易日累计开仓:{int(opens_today)}(硬上限 {int(hard_limit)},提醒 {int(alert_threshold)})"
return f"本交易日累计开仓:{int(opens_today)}(提醒阈值 {int(alert_threshold)})"
lib/trade/entry_model_lib.py
+6 -3
View File
@@ -441,10 +441,13 @@ def enrich_entry_model_display(item: dict) -> dict:
return item
def open_position_button_label(policy: TradePolicy, sizing_mode: str) -> str:
from lib.trade.position_sizing_lib import mode_label_zh
def _sizing_mode_label_zh(mode: str) -> str:
m = (mode or "").strip().lower()
return {"risk": "按风险", "full_margin": "满保证金"}.get(m, m or "-")
mode_txt = mode_label_zh(sizing_mode)
def open_position_button_label(policy: TradePolicy, sizing_mode: str) -> str:
mode_txt = _sizing_mode_label_zh(sizing_mode)
if is_intraday_trading_profile(policy):
return f"开仓(日内·{mode_txt})"
return f"开仓({mode_txt})"
lib/trade/force_close_lib.py
-366
View File
@@ -1,366 +0,0 @@
"""整点强制清仓(FORCE_CLOSE_*):UI 标识与持仓倒计时."""
from __future__ import annotations
import os
import time
from datetime import datetime, timedelta
from typing import Any, Optional
from zoneinfo import ZoneInfo
FORCE_CLOSE_RESULT = "强制清仓"
# 默认宽限(分钟);运行时优先读 FORCE_CLOSE_GRACE_MINUTES
FORCE_CLOSE_GRACE_MINUTES = 5
def force_close_grace_minutes(override: Any = None) -> int:
"""整点强制清仓执行窗口长度(分钟)."""
if override is not None and str(override).strip() != "":
raw = override
else:
raw = os.getenv("FORCE_CLOSE_GRACE_MINUTES")
if raw is None or str(raw).strip() == "":
raw = FORCE_CLOSE_GRACE_MINUTES
try:
return max(1, int(raw))
except (TypeError, ValueError):
return max(1, int(FORCE_CLOSE_GRACE_MINUTES))
def app_timezone_name() -> str:
return (os.getenv("APP_TIMEZONE") or os.getenv("TZ") or "Asia/Shanghai").strip() or "Asia/Shanghai"
def normalize_force_close_bj_hour(value: Any) -> int:
try:
h = int(value)
except (TypeError, ValueError):
return 0
return max(0, min(23, h))
def _now_dt(*, now_ms: Optional[int] = None, tz_name: Optional[str] = None) -> datetime:
tz = ZoneInfo(tz_name or app_timezone_name())
if now_ms is None:
return datetime.now(tz)
return datetime.fromtimestamp(int(now_ms) / 1000, tz=tz)
def force_close_hour_label(bj_hour: Any) -> str:
return f"{normalize_force_close_bj_hour(bj_hour):02d}:00"
def force_close_label(bj_hour: Any) -> str:
return f"强制清仓 {force_close_hour_label(bj_hour)}"
def is_force_close_active_hour(
bj_hour: Any,
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
grace_minutes: Optional[int] = None,
) -> bool:
"""当前是否处于整点强制清仓执行窗口(整点起 grace 分钟内)."""
return is_force_close_executing(
bj_hour,
now_ms=now_ms,
tz_name=tz_name,
grace_minutes=grace_minutes,
)
def is_force_close_executing(
bj_hour: Any,
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
grace_minutes: Optional[int] = None,
) -> bool:
hour = normalize_force_close_bj_hour(bj_hour)
now = _now_dt(now_ms=now_ms, tz_name=tz_name)
target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
if now < target:
return False
grace = force_close_grace_minutes(grace_minutes)
end = target + timedelta(minutes=grace)
return now < end
def force_close_blocks_new_open(
enabled: bool,
bj_hour: Any,
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
grace_minutes: Optional[int] = None,
) -> tuple[bool, str]:
"""强制清仓执行窗口内禁止新开仓.返回 (是否拦截, 说明文案)."""
if not enabled:
return False, ""
grace = force_close_grace_minutes(grace_minutes)
if not is_force_close_executing(
bj_hour, now_ms=now_ms, tz_name=tz_name, grace_minutes=grace
):
return False, ""
label = force_close_hour_label(bj_hour)
return (
True,
f"强制清仓窗口内(北京时间 {label} 起 {grace} 分钟),暂不可开仓",
)
def parse_closed_at_dt(
closed_at: Any,
*,
tz_name: Optional[str] = None,
) -> Optional[datetime]:
if closed_at is None:
return None
text = str(closed_at).strip()
if not text:
return None
tz = ZoneInfo(tz_name or app_timezone_name())
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"):
try:
return datetime.strptime(text, fmt).replace(tzinfo=tz)
except ValueError:
continue
return None
def is_close_at_force_close_window(
closed_at: Any,
bj_hour: Any,
*,
grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> bool:
"""平仓时刻是否落在北京时间整点强制清仓窗口内."""
dt = parse_closed_at_dt(closed_at, tz_name=tz_name)
if dt is None:
return False
hour = normalize_force_close_bj_hour(bj_hour)
if dt.hour != hour:
return False
return dt.minute < force_close_grace_minutes(grace_minutes)
def infer_force_close_result(
closed_at: Any,
*,
enabled: bool,
bj_hour: Any,
grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> Optional[str]:
if not enabled:
return None
if is_close_at_force_close_window(
closed_at, bj_hour, grace_minutes=grace_minutes, tz_name=tz_name
):
return FORCE_CLOSE_RESULT
return None
def coerce_force_close_result(
result: Optional[str],
closed_at: Any,
*,
enabled: bool,
bj_hour: Any,
miss_reason: Optional[str] = None,
grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> tuple[str, str]:
"""同步平仓归类:整点窗口内优先记为强制清仓."""
res = (result or "").strip()
note = (miss_reason or "").strip()
if res == FORCE_CLOSE_RESULT:
return res, note
fc = infer_force_close_result(
closed_at,
enabled=enabled,
bj_hour=bj_hour,
grace_minutes=grace_minutes,
tz_name=tz_name,
)
if not fc:
return res, note
if not note:
note = f"北京时间 {force_close_hour_label(bj_hour)} 整点风控清仓"
return fc, note
def apply_force_close_display_result(
result: Optional[str],
closed_at: Any,
*,
enabled: bool,
bj_hour: Any,
grace_minutes: Optional[int] = None,
tz_name: Optional[str] = None,
) -> str:
"""展示层:外部平仓/手动平仓若落在整点窗口,显示为强制清仓."""
res = (result or "").strip()
if res == FORCE_CLOSE_RESULT:
return res
fc = infer_force_close_result(
closed_at,
enabled=enabled,
bj_hour=bj_hour,
grace_minutes=grace_minutes,
tz_name=tz_name,
)
if fc and (res in ("", "外部平仓", "手动平仓") or res.startswith("外部平仓")):
return fc
return res
def compute_next_force_close_at_ms(
*,
bj_hour: Any,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
) -> Optional[int]:
"""下一次强制清仓时刻(北京时间整点)的 epoch 毫秒."""
hour = normalize_force_close_bj_hour(bj_hour)
now = _now_dt(now_ms=now_ms, tz_name=tz_name)
target = now.replace(hour=hour, minute=0, second=0, microsecond=0)
if now >= target:
target += timedelta(days=1)
return int(target.timestamp() * 1000)
def force_close_remaining_seconds(
close_at_ms: Any,
*,
now_ms: Optional[int] = None,
) -> Optional[int]:
try:
close_at = int(close_at_ms)
except (TypeError, ValueError):
return None
now = int(now_ms if now_ms is not None else time.time() * 1000)
return max(0, int((close_at - now) / 1000))
def format_force_close_countdown(seconds: Any, *, active: bool = False) -> str:
if active:
return "执行中"
try:
sec = max(0, int(seconds))
except (TypeError, ValueError):
return "--:--:--"
h = sec // 3600
m = (sec % 3600) // 60
s = sec % 60
return f"{h:02d}:{m:02d}:{s:02d}"
def build_force_close_state(
enabled: bool,
bj_hour: Any,
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
has_active_positions: Optional[bool] = None,
) -> dict[str, Any]:
"""实例级强制清仓状态(模板 / API 共用)."""
grace = force_close_grace_minutes()
if not enabled:
return {
"enabled": False,
"bj_hour": normalize_force_close_bj_hour(bj_hour),
"hour_label": force_close_hour_label(bj_hour),
"label": force_close_label(bj_hour),
"grace_minutes": grace,
"next_at_ms": None,
"remaining_sec": None,
"countdown": "",
"active": False,
"executing": False,
}
hour = normalize_force_close_bj_hour(bj_hour)
executing = is_force_close_executing(
hour, now_ms=now_ms, tz_name=tz_name, grace_minutes=grace
)
active = executing and (has_active_positions is not False)
next_at_ms = compute_next_force_close_at_ms(bj_hour=hour, now_ms=now_ms, tz_name=tz_name)
rem = force_close_remaining_seconds(next_at_ms, now_ms=now_ms) if next_at_ms else None
return {
"enabled": True,
"bj_hour": hour,
"hour_label": force_close_hour_label(hour),
"label": force_close_label(hour),
"grace_minutes": grace,
"next_at_ms": next_at_ms,
"remaining_sec": rem,
"countdown": format_force_close_countdown(rem, active=active),
"active": active,
"executing": executing,
}
def force_close_template_context(
enabled: bool,
bj_hour: Any,
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
has_active_positions: Optional[bool] = None,
) -> dict[str, dict[str, Any]]:
return {
"force_close": build_force_close_state(
enabled,
bj_hour,
now_ms=now_ms,
tz_name=tz_name,
has_active_positions=has_active_positions,
)
}
def apply_force_close_to_payload(
payload: dict[str, Any],
*,
enabled: bool,
bj_hour: Any,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
) -> None:
"""为 active 持仓 JSON 附加整点强制清仓倒计时."""
state = build_force_close_state(
enabled,
bj_hour,
now_ms=now_ms,
tz_name=tz_name,
has_active_positions=True,
)
payload["force_close_enabled"] = bool(state["enabled"])
payload["force_close_bj_hour"] = state["bj_hour"]
payload["force_close_at_ms"] = state["next_at_ms"]
payload["force_close_label"] = state["label"] if state["enabled"] else ""
payload["force_close_remaining_sec"] = state["remaining_sec"]
payload["force_close_countdown"] = state["countdown"]
payload["force_close_active"] = bool(state["active"])
def enrich_orders_force_close(
orders: list[dict[str, Any]],
enabled: bool,
bj_hour: Any,
*,
now_ms: Optional[int] = None,
tz_name: Optional[str] = None,
) -> None:
if not enabled or not orders:
return
for item in orders:
if isinstance(item, dict):
apply_force_close_to_payload(
item,
enabled=enabled,
bj_hour=bj_hour,
now_ms=now_ms,
tz_name=tz_name,
)
lib/trade/manual_sltp_lib.py
-136
View File
@@ -1,136 +0,0 @@
"""实盘人工下单:止盈止损模式(价格 / 百分比 / 固定盈亏比)."""
from __future__ import annotations
from typing import Any, Optional, Tuple
MANUAL_FIXED_RR_DEFAULT = 1.5
SLTP_MODE_PRICE = "price"
SLTP_MODE_PCT = "pct"
SLTP_MODE_FIXED_RR = "fixed_rr"
OPEN_SLTP_MODES = frozenset({SLTP_MODE_PRICE, SLTP_MODE_PCT, SLTP_MODE_FIXED_RR})
ENTRUST_SLTP_MODES = frozenset({SLTP_MODE_PRICE, SLTP_MODE_PCT})
def normalize_open_sltp_mode(raw: Optional[str]) -> str:
mode = (raw or SLTP_MODE_FIXED_RR).strip().lower()
if mode in OPEN_SLTP_MODES:
return mode
return SLTP_MODE_PRICE
def normalize_entrust_sltp_mode(raw: Optional[str]) -> str:
mode = (raw or SLTP_MODE_PRICE).strip().lower()
if mode in ENTRUST_SLTP_MODES:
return mode
return SLTP_MODE_PRICE
def parse_fixed_rr(raw: Any, *, default: float = MANUAL_FIXED_RR_DEFAULT) -> float:
try:
v = float(raw)
if v > 0:
return v
except (TypeError, ValueError):
pass
return float(default)
def calc_tp_from_fixed_rr(
direction: str,
entry_price: float,
stop_loss: float,
rr_ratio: float,
) -> float:
entry = float(entry_price)
sl = float(stop_loss)
rr = float(rr_ratio)
if entry <= 0 or sl <= 0 or rr <= 0:
raise ValueError("固定盈亏比参数无效")
side = (direction or "long").strip().lower()
if side == "short":
risk = sl - entry
if risk <= 0:
raise ValueError("止损方向不合法:做空时止损须高于入场价")
return entry - risk * rr
risk = entry - sl
if risk <= 0:
raise ValueError("止损方向不合法:做多时止损须低于入场价")
return entry + risk * rr
def _resolve_pct_sltp(direction: str, live_price: float, data: dict[str, Any]) -> Tuple[float, float]:
sl_pct = float(data.get("sl_pct") or 0)
tp_pct = float(data.get("tp_pct") or 0)
if sl_pct <= 0 or tp_pct <= 0:
raise ValueError("百分比止盈止损须为正数")
sl_ratio = sl_pct / 100.0
tp_ratio = tp_pct / 100.0
entry = float(live_price)
if (direction or "long").strip().lower() == "short":
stop_loss = entry * (1 + sl_ratio)
take_profit = entry * (1 - tp_ratio)
else:
stop_loss = entry * (1 - sl_ratio)
take_profit = entry * (1 + tp_ratio)
return stop_loss, take_profit
def _resolve_price_sltp(
data: dict[str, Any],
*,
fallback_sl: Optional[float] = None,
fallback_tp: Optional[float] = None,
require_tp: bool = True,
) -> Tuple[float, float]:
stop_loss = float(data.get("sl") or data.get("stop_loss") or 0)
take_profit = float(data.get("tp") or data.get("take_profit") or data.get("tgt") or 0)
if stop_loss <= 0 and fallback_sl is not None:
stop_loss = float(fallback_sl)
if take_profit <= 0 and fallback_tp is not None:
take_profit = float(fallback_tp)
if stop_loss <= 0:
raise ValueError("止损价格须大于 0" if require_tp else "请填写止损价格")
if require_tp and take_profit <= 0:
raise ValueError("止盈止损价格须大于 0" if fallback_tp is None else "请填写止盈价格,或保留原计划止盈")
return stop_loss, take_profit
def resolve_open_sltp_prices(
direction: str,
live_price: float,
sltp_mode: Optional[str],
data: dict[str, Any],
) -> Tuple[float, float]:
"""新开仓 /add_order:支持 price,pct,fixed_rr."""
mode = normalize_open_sltp_mode(sltp_mode)
if mode == SLTP_MODE_PCT:
return _resolve_pct_sltp(direction, live_price, data)
if mode == SLTP_MODE_FIXED_RR:
stop_loss, _ = _resolve_price_sltp(data, require_tp=False)
rr = parse_fixed_rr(data.get("fixed_rr"))
take_profit = calc_tp_from_fixed_rr(direction, live_price, stop_loss, rr)
return stop_loss, take_profit
return _resolve_price_sltp(data, require_tp=True)
def resolve_entrust_sltp_prices(
direction: str,
live_price: float,
sltp_mode: Optional[str],
data: dict[str, Any],
*,
fallback_sl: Optional[float] = None,
fallback_tp: Optional[float] = None,
) -> Tuple[float, float]:
"""持仓委托弹窗:仅 price / pct,不校验盈亏比."""
mode = normalize_entrust_sltp_mode(sltp_mode)
if mode == SLTP_MODE_PCT:
return _resolve_pct_sltp(direction, live_price, data)
return _resolve_price_sltp(
data,
fallback_sl=fallback_sl,
fallback_tp=fallback_tp,
require_tp=True,
)
lib/trade/open_trade_gate_lib.py
-57
View File
@@ -1,57 +0,0 @@
"""三所开仓门禁:账户风控 + 强制清仓窗口 + 仓位/日开仓上限."""
from __future__ import annotations
from typing import Any, Optional
from lib.trade.daily_open_limit_lib import can_trade_new_open
from lib.trade.force_close_lib import force_close_blocks_new_open
def resolve_manual_open_gate(
*,
time_allows: bool,
active_count: int,
max_active_positions: int,
opens_today: int,
hard_limit: int,
risk_status: Optional[dict[str, Any]],
force_close_enabled: bool,
force_close_bj_hour: Any,
now_ms: Optional[int] = None,
reset_hour: int = 8,
) -> dict[str, Any]:
"""汇总是否可开仓及按钮旁说明文案."""
rs = risk_status if isinstance(risk_status, dict) else {}
risk_can = bool(rs.get("can_trade", True))
fc_block, fc_note = force_close_blocks_new_open(
bool(force_close_enabled),
force_close_bj_hour,
now_ms=now_ms,
)
can_trade = can_trade_new_open(
time_allows=time_allows,
active_count=active_count,
max_active_positions=max_active_positions,
opens_today=opens_today,
hard_limit=hard_limit,
extra_blocks=(not risk_can) or fc_block,
)
note = ""
if fc_block and fc_note:
note = fc_note
elif not risk_can:
note = str(rs.get("reason") or "账户冷静期/日冻结中,暂不可开仓")
elif not time_allows:
note = f"未到北京时间 {int(reset_hour)}:00,暂不可开仓"
elif int(active_count) >= int(max_active_positions):
note = f"已达最大持仓数({int(active_count)}/{int(max_active_positions)}),暂不可开仓"
elif int(hard_limit) > 0 and int(opens_today) >= int(hard_limit):
note = (
f"本交易日开仓已达上限({int(opens_today)}/{int(hard_limit)}),"
f"次日北京时间 {int(reset_hour)}:00 后恢复"
)
return {
"can_trade": can_trade,
"open_block_note": note if not can_trade else "",
"force_close_blocks": fc_block,
}
lib/trade/order_monitor_display_lib.py
-461
View File
@@ -1,461 +0,0 @@
"""实时持仓展示:开仓快照盈亏比,交易所止损是否已保本."""
from __future__ import annotations
from typing import Any, Callable, Optional
def _positive_float(value: Any) -> Optional[float]:
try:
v = float(value)
return v if v > 0 else None
except (TypeError, ValueError):
return None
def snapshot_stop_loss(initial_stop_loss: Any, stop_loss: Any) -> Optional[float]:
"""展示盈亏比 / 交易记录时优先用开仓时止损快照,不用后续改单后的止损."""
sl = _positive_float(initial_stop_loss)
if sl is not None:
return sl
return _positive_float(stop_loss)
def monitor_open_stop_loss(row: Any) -> Optional[float]:
"""从 order_monitors 行取开仓止损快照."""
try:
keys = row.keys() if hasattr(row, "keys") else ()
except Exception:
keys = ()
init = row["initial_stop_loss"] if "initial_stop_loss" in keys else None
cur = row["stop_loss"] if "stop_loss" in keys else None
if init is None and isinstance(row, dict):
init = row.get("initial_stop_loss")
cur = row.get("stop_loss")
return snapshot_stop_loss(init, cur)
def snapshot_rr(
calc_rr_ratio_fn: Callable[..., Optional[float]],
direction: str,
trigger_price: Any,
initial_stop_loss: Any,
stop_loss: Any,
take_profit: Any,
) -> Optional[float]:
entry = _positive_float(trigger_price)
sl = snapshot_stop_loss(initial_stop_loss, stop_loss)
tp = _positive_float(take_profit)
if entry is None or sl is None or tp is None:
return None
return calc_rr_ratio_fn(direction or "long", entry, sl, tp)
def tpsl_slot_trigger_price(slot: Any) -> Optional[float]:
if not isinstance(slot, dict):
return None
for key in ("trigger_price", "trigger_display"):
v = _positive_float(slot.get(key))
if v is not None:
return v
return None
def stop_is_profit_protecting(direction: str, entry_price: Any, stop_loss: Any) -> bool:
"""
止损是否已在盈利侧(保本/锁盈),不再适用「开仓盈亏比」风控.
做空:止损 < 成交价;做多:止损 > 成交价.
"""
entry = _positive_float(entry_price)
sl = _positive_float(stop_loss)
if entry is None or sl is None:
return False
d = (direction or "long").strip().lower()
if d == "short":
return sl < entry
return sl > entry
def tpsl_update_passes_rr_gate(
direction: str,
entry_price: Any,
stop_loss: Any,
take_profit: Any,
min_rr: float,
calc_rr_ratio_fn: Callable[..., Optional[float]],
) -> tuple[bool, Optional[str]]:
"""持仓委托改价:盈利侧止损跳过最低盈亏比;否则按开仓价几何校验."""
if stop_is_profit_protecting(direction, entry_price, stop_loss):
return True, None
rr = calc_rr_ratio_fn(direction or "long", entry_price, stop_loss, take_profit)
if rr is not None and rr >= float(min_rr):
return True, None
rr_txt = f"{rr:.4f}" if rr is not None else "无法计算"
return False, f"计划盈亏比 {rr_txt}:1 低于最低要求 {min_rr}:1(盈利侧保本止损不受此限)"
def resolve_breakeven_entry_price(entry_price: Any, avg_entry_price: Any = None) -> Optional[float]:
"""保本判断基准价:有持仓加权均价时优先(滚仓后),否则用首仓成交价."""
avg = _positive_float(avg_entry_price)
if avg is not None:
return avg
return _positive_float(entry_price)
def stale_breakeven_armed(direction: str, entry_price: Any, stop_loss: Any, breakeven_armed: Any) -> bool:
"""止损已回到亏损侧时 breakeven_armed 视为过期(如滚仓下移止损)."""
try:
armed = int(breakeven_armed or 0) != 0
except (TypeError, ValueError):
return False
if not armed:
return False
return not stop_is_profit_protecting(direction, entry_price, stop_loss)
def is_sl_breakeven_secured(direction: str, entry_price: Any, exchange_sl_price: Any) -> bool:
"""
交易所当前止损相对开仓成交价是否已保本.
做多:止损 >= 成交价;做空:止损 <= 成交价.
"""
entry = _positive_float(entry_price)
sl = _positive_float(exchange_sl_price)
if entry is None or sl is None:
return False
d = (direction or "long").strip().lower()
if d == "short":
return sl <= entry
return sl >= entry
def sl_breakeven_from_exchange_tpsl(
direction: str,
entry_price: Any,
exchange_tpsl: Any,
) -> bool:
if not isinstance(exchange_tpsl, dict):
return False
sl_px = tpsl_slot_trigger_price(exchange_tpsl.get("sl"))
if sl_px is None:
return False
return is_sl_breakeven_secured(direction, entry_price, sl_px)
def enrich_order_display_fields(item: dict[str, Any], calc_rr_ratio_fn: Callable[..., Optional[float]]) -> dict[str, Any]:
item["rr_ratio"] = snapshot_rr(
calc_rr_ratio_fn,
item.get("direction") or "long",
item.get("trigger_price"),
item.get("initial_stop_loss"),
item.get("stop_loss"),
item.get("take_profit"),
)
return item
def apply_order_live_price_display(
payload: dict[str, Any],
symbol: Any,
ticker_price: Any,
exchange_mark_price: Any,
format_price_fn: Callable[[Any, Any], str],
) -> dict[str, Any]:
"""标记价/现价展示:与交易所 price_to_precision 对齐,避免前端 toFixed(8)."""
px_for_fmt = ticker_price
mark_raw = exchange_mark_price
if mark_raw is not None:
try:
px_for_fmt = float(mark_raw)
except (TypeError, ValueError):
pass
px_disp = format_price_fn(symbol, px_for_fmt)
payload["price_display"] = px_disp
if mark_raw is not None:
try:
payload["exchange_mark_price_display"] = format_price_fn(symbol, float(mark_raw))
except (TypeError, ValueError):
payload["exchange_mark_price_display"] = px_disp
else:
payload["exchange_mark_price_display"] = None
return payload
def resolve_live_tpsl_prices(
plan_sl: Any,
plan_tp: Any,
exchange_tpsl: Any,
) -> tuple[Optional[float], Optional[float], Optional[float], Optional[float]]:
"""返回 (展示用止损, 展示用止盈, 交易所止损, 交易所止盈)."""
ex_sl = ex_tp = None
if isinstance(exchange_tpsl, dict):
ex_sl = tpsl_slot_trigger_price(exchange_tpsl.get("sl"))
ex_tp = tpsl_slot_trigger_price(exchange_tpsl.get("tp"))
disp_sl = ex_sl if ex_sl is not None else _positive_float(plan_sl)
disp_tp = ex_tp if ex_tp is not None else _positive_float(plan_tp)
return disp_sl, disp_tp, ex_sl, ex_tp
def calc_risk_fraction(direction: str, entry_price: Any, stop_loss: Any) -> Optional[float]:
"""|入场-止损|/入场;盈利侧止损返回 0."""
entry = _positive_float(entry_price)
sl = _positive_float(stop_loss)
if entry is None or sl is None:
return None
d = (direction or "long").strip().lower()
if d == "short":
risk = sl - entry
else:
risk = entry - sl
if risk <= 0:
return 0.0
return risk / entry
def calc_latest_risk_amount(
direction: str,
entry_price: Any,
stop_loss: Any,
*,
margin_capital: Any = None,
leverage: Any = None,
exchange_notional: Any = None,
contracts: Any = None,
contract_size: Any = None,
mark_price: Any = None,
funds_decimals: int = 2,
) -> Optional[float]:
"""按当前止损与持仓名义价值估算最新风险(U)."""
rf = calc_risk_fraction(direction, entry_price, stop_loss)
if rf is None:
return None
if rf <= 0:
return 0.0
notional = _positive_float(exchange_notional)
if notional is None:
try:
mc = float(margin_capital or 0)
lev = float(leverage or 0)
if mc > 0 and lev > 0:
notional = mc * lev
except (TypeError, ValueError):
pass
if notional is None:
try:
c = abs(float(contracts or 0))
cs = float(contract_size or 1)
if cs <= 0:
cs = 1.0
px = _positive_float(mark_price) or _positive_float(entry_price)
if c > 0 and px is not None:
notional = c * cs * px
except (TypeError, ValueError):
pass
if notional is None or notional <= 0:
return None
return round(notional * rf, funds_decimals)
def order_monitor_tpsl_needs_sync(
plan_sl: Any,
plan_tp: Any,
exchange_tpsl: Any,
*,
eps: float = 1e-12,
) -> tuple[Optional[float], Optional[float], bool]:
"""若交易所 TP/SL 与库中不一致,返回应写回的 (sl, tp) 及是否需更新."""
_, _, ex_sl, ex_tp = resolve_live_tpsl_prices(plan_sl, plan_tp, exchange_tpsl)
try:
cur_sl = float(plan_sl or 0)
cur_tp = float(plan_tp or 0)
except (TypeError, ValueError):
cur_sl, cur_tp = 0.0, 0.0
new_sl = ex_sl if ex_sl is not None else cur_sl
new_tp = ex_tp if ex_tp is not None else cur_tp
changed = (
(ex_sl is not None and abs(new_sl - cur_sl) > eps)
or (ex_tp is not None and abs(new_tp - cur_tp) > eps)
)
return new_sl, new_tp, changed
def apply_order_price_display_fields(
payload: dict[str, Any],
*,
direction: str,
entry_price: Any,
initial_stop_loss: Any,
stop_loss: Any,
take_profit: Any,
calc_rr_ratio_fn: Callable[..., Optional[float]],
exchange_tpsl: Any = None,
format_price_fn: Optional[Callable[[Any, Any], str]] = None,
symbol: Any = None,
margin_capital: Any = None,
leverage: Any = None,
exchange_notional: Any = None,
contracts: Any = None,
contract_size: Any = None,
mark_price: Any = None,
avg_entry_price: Any = None,
funds_decimals: int = 2,
) -> dict[str, Any]:
disp_sl, disp_tp, _, _ = resolve_live_tpsl_prices(stop_loss, take_profit, exchange_tpsl)
payload["stop_loss_raw"] = _positive_float(stop_loss)
payload["take_profit_raw"] = _positive_float(take_profit)
payload["rr_ratio"] = snapshot_rr(
calc_rr_ratio_fn,
direction,
entry_price,
initial_stop_loss,
stop_loss,
take_profit,
)
risk_entry = resolve_breakeven_entry_price(entry_price, avg_entry_price)
payload["avg_entry_price"] = risk_entry
payload["sl_breakeven_secured"] = sl_breakeven_from_exchange_tpsl(
direction, risk_entry, exchange_tpsl
)
payload["stop_loss"] = disp_sl
payload["take_profit"] = disp_tp
if disp_sl is not None and disp_tp is not None:
payload["display_rr_ratio"] = calc_rr_ratio_fn(
direction or "long", entry_price, disp_sl, disp_tp
)
else:
payload["display_rr_ratio"] = None
if contracts is not None:
try:
from lib.market.position_metrics_lib import normalize_contracts_qty
c = normalize_contracts_qty(contracts)
if c > 0:
payload["contracts"] = c
except (TypeError, ValueError):
pass
payload["latest_risk_amount"] = calc_latest_risk_amount(
direction,
risk_entry,
disp_sl if disp_sl is not None else stop_loss,
margin_capital=margin_capital,
leverage=leverage,
exchange_notional=exchange_notional,
contracts=payload.get("contracts") if payload.get("contracts") is not None else contracts,
contract_size=contract_size,
mark_price=mark_price,
funds_decimals=funds_decimals,
)
tp_for_reward = disp_tp if disp_tp is not None else _positive_float(take_profit)
qty_for_reward = payload.get("contracts")
if qty_for_reward is None and contracts is not None:
try:
qty_for_reward = abs(float(contracts))
except (TypeError, ValueError):
qty_for_reward = None
if risk_entry is not None and tp_for_reward is not None and qty_for_reward:
try:
def reward_at_tp_usdt(direction, entry, tp, contracts, contract_size, fee_rate=0.0):
try:
d = (direction or "long").lower()
e, t, c, cs = float(entry), float(tp), float(contracts), float(contract_size or 1)
if e <= 0 or t <= 0 or c <= 0:
return None
raw = (t - e) * c * cs if d == "long" else (e - t) * c * cs
return round(raw, 2)
except Exception:
return None
reward = reward_at_tp_usdt(
direction,
risk_entry,
tp_for_reward,
float(qty_for_reward),
contract_size=float(contract_size or 1.0),
)
payload["reward_at_tp_usdt"] = (
round(reward, funds_decimals) if reward is not None else None
)
except Exception:
payload["reward_at_tp_usdt"] = None
else:
payload["reward_at_tp_usdt"] = None
if format_price_fn is not None and symbol is not None:
payload["stop_loss_display"] = (
format_price_fn(symbol, disp_sl) if disp_sl is not None else "—"
)
payload["take_profit_display"] = (
format_price_fn(symbol, disp_tp) if disp_tp is not None else "—"
)
mark_raw = mark_price if mark_price is not None else None
if mark_raw is not None and format_price_fn is not None and symbol is not None:
try:
payload["exchange_mark_price_display"] = format_price_fn(symbol, float(mark_raw))
except (TypeError, ValueError):
payload["exchange_mark_price_display"] = None
return payload
def enrich_active_monitor_tpsl_json(
row: Any,
stop_loss: Any,
take_profit: Any,
exchange_tpsl: Any,
*,
position_row: Any = None,
exchange_notional: Any = None,
contracts: Any = None,
contract_size: float = 1.0,
mark_price: Any = None,
calc_rr_ratio_fn: Callable[..., Optional[float]],
format_price_fn: Optional[Callable[[Any, Any], str]] = None,
symbol: Any = None,
funds_decimals: int = 2,
) -> dict[str, Any]:
"""place_tpsl 响应:展示用 TP/SL,最新风险,当前盈亏比."""
def _row_val(key: str, default=None):
try:
if hasattr(row, "keys") and key in row.keys():
return row[key]
except Exception:
pass
if isinstance(row, dict):
return row.get(key, default)
return default
direction = _row_val("direction") or "long"
entry = _row_val("trigger_price")
init_sl = _row_val("initial_stop_loss")
margin = _row_val("margin_capital")
leverage = _row_val("leverage")
if position_row is not None:
from lib.market.position_metrics_lib import parse_position_entry_price, position_contracts
live_c = position_contracts(position_row)
if abs(live_c) >= 1e-12:
contracts = abs(live_c)
avg_entry = parse_position_entry_price(position_row)
else:
avg_entry = None
payload: dict[str, Any] = {
"stop_loss": stop_loss,
"take_profit": take_profit,
}
apply_order_price_display_fields(
payload,
direction=direction,
entry_price=entry,
initial_stop_loss=init_sl,
stop_loss=stop_loss,
take_profit=take_profit,
calc_rr_ratio_fn=calc_rr_ratio_fn,
exchange_tpsl=exchange_tpsl,
format_price_fn=format_price_fn,
symbol=symbol or _row_val("symbol"),
margin_capital=margin,
leverage=leverage,
exchange_notional=exchange_notional,
contracts=contracts,
contract_size=contract_size,
mark_price=mark_price,
avg_entry_price=avg_entry,
funds_decimals=funds_decimals,
)
return payload
lib/trade/position_sizing_lib.py
-135
View File
@@ -1,135 +0,0 @@
"""
三所共用:计仓模式 risk(以损定仓)| full_margin(全仓杠杆).
仅 env POSITION_SIZING_MODE 切换;须无持仓(由部署流程保证).
"""
from __future__ import annotations
import os
from typing import Any, Optional, Tuple
MODE_RISK = "risk"
MODE_FULL_MARGIN = "full_margin"
VALID_MODES = frozenset({MODE_RISK, MODE_FULL_MARGIN})
OPEN_SOURCE_MANUAL = "manual"
# 历史兼容常量(策略/关键位自动单已移除,assert 仍可引用)
OPEN_SOURCE_KEY_AUTO = "key_auto"
OPEN_SOURCE_KEY_FIB = "key_fib"
OPEN_SOURCE_KEY_TRIGGER = "key_trigger"
OPEN_SOURCE_TREND = "trend"
OPEN_SOURCE_ROLL = "roll"
FULL_MARGIN_BLOCKED_SOURCES = frozenset()
def normalize_position_sizing_mode(raw: Optional[str]) -> str:
v = (raw or MODE_RISK).strip().lower()
if v in ("full", "full_margin", "fullmargin", "全仓", "全仓杠杆"):
return MODE_FULL_MARGIN
return MODE_RISK if v in ("risk", "r", "以损定仓", "") else MODE_RISK
def load_position_sizing_mode(env: Optional[dict] = None) -> str:
e = env if env is not None else os.environ
return normalize_position_sizing_mode(e.get("POSITION_SIZING_MODE"))
def is_full_margin_mode(mode: str) -> bool:
return normalize_position_sizing_mode(mode) == MODE_FULL_MARGIN
def mode_label_zh(mode: str) -> str:
return "全仓杠杆" if is_full_margin_mode(mode) else "以损定仓"
def leverage_for_full_margin(symbol: str, btc_leverage: int, alt_leverage: int) -> int:
sym = (symbol or "").strip().upper()
if sym.startswith("BTC") or sym.startswith("ETH"):
return max(1, int(btc_leverage or 10))
return max(1, int(alt_leverage or 5))
def round_funds(value: float, decimals: int = 2) -> float:
return round(float(value), int(decimals))
def risk_percent_for_storage(mode: str, risk_percent: float) -> Optional[float]:
"""全仓杠杆:库内不写风险百分比(仅 risk_amount U)."""
if is_full_margin_mode(mode):
return None
return risk_percent
def format_risk_display_text(
mode: str,
risk_percent: Optional[float],
risk_amount: Optional[float],
*,
decimals: int = 2,
) -> str:
"""持仓/通知「风险」文案:全仓仅 U;以损定仓为 %≈U."""
amt: Optional[float] = None
if risk_amount is not None and risk_amount != "":
try:
amt = float(risk_amount)
except (TypeError, ValueError):
amt = None
if is_full_margin_mode(mode):
if amt is None:
return "—"
return f"{round_funds(amt, decimals)}U"
pct: Optional[float] = None
if risk_percent is not None and risk_percent != "":
try:
pct = float(risk_percent)
except (TypeError, ValueError):
pct = None
pct_txt = f"{pct:g}" if pct is not None else "—"
amt_txt = round_funds(amt, decimals) if amt is not None else "—"
return f"{pct_txt}%≈{amt_txt}U"
def assert_open_source_allowed(mode: str, source: str) -> Tuple[bool, str]:
if not is_full_margin_mode(mode):
return True, ""
src = (source or "").strip().lower()
if src in FULL_MARGIN_BLOCKED_SOURCES:
return False, (
"当前为全仓杠杆模式(POSITION_SIZING_MODE=full_margin),"
"不允许关键位突破/斐波自动开仓,趋势回调与顺势加仓;"
"仅支持实盘人工下单与阻力/支撑提醒."
)
return True, ""
def full_margin_requires_flat_position(active_count: int) -> Tuple[bool, str]:
if active_count > 0:
return False, "全仓杠杆模式仅允许单仓且无其它持仓,请先平仓后再开仓"
return True, ""
def compute_full_margin_sizing(
*,
symbol: str,
available_usdt: float,
capital_base: float,
buffer_ratio: float,
btc_leverage: int,
alt_leverage: int,
funds_decimals: int = 2,
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
if available_usdt is None or float(available_usdt) <= 0:
return None, "全仓杠杆:无法读取合约账户可用保证金"
lev = leverage_for_full_margin(symbol, btc_leverage, alt_leverage)
margin = round_funds(float(available_usdt) * float(buffer_ratio), funds_decimals)
if margin <= 0:
return None, "全仓杠杆:可用保证金不足"
notional = round_funds(margin * lev, funds_decimals)
ratio = round(margin / float(capital_base) * 100, 2) if capital_base else 0.0
return {
"margin_capital": margin,
"leverage": lev,
"notional_value": notional,
"position_ratio": ratio,
"mode": MODE_FULL_MARGIN,
}, None
lib/trade/trade_exchange_stats_lib.py
-229
View File
@@ -1,229 +0,0 @@
"""平仓交易:交易所口径双边成交额与手续费(三所共用聚合逻辑)."""
from __future__ import annotations
from typing import Any, Callable, Optional
def _coerce_ts_ms(raw: Any) -> int | None:
if raw in (None, ""):
return None
try:
v = int(raw)
return v if v > 1_000_000_000_000 else v * 1000
except (TypeError, ValueError):
return None
def quote_turnover_usdt_from_fill(trade: dict, *, contract_size: float = 1.0) -> float:
"""单笔成交的报价币成交额(USDT 口径)."""
info = trade.get("info") or {}
if not isinstance(info, dict):
info = {}
for key in ("quoteQty", "quote_qty", "fillNotionalUsd", "notional"):
try:
v = float(info.get(key) or 0)
if v > 0:
return abs(v)
except (TypeError, ValueError):
continue
try:
cost = float(trade.get("cost") or 0)
if cost > 0:
return abs(cost)
except (TypeError, ValueError):
pass
try:
price = float(trade.get("price") or 0)
amount = float(trade.get("amount") or 0) * float(contract_size or 1.0)
if price > 0 and amount > 0:
return abs(price * amount)
except (TypeError, ValueError):
pass
return 0.0
def commission_usdt_from_fill(trade: dict) -> float:
"""单笔成交手续费(正数表示成本)."""
fee = trade.get("fee")
if isinstance(fee, dict):
try:
cost = float(fee.get("cost") or 0)
except (TypeError, ValueError):
cost = 0.0
if cost != 0:
cur = str(fee.get("currency") or "USDT").upper()
if cur in ("USDT", "USD", "BUSD", "USDC"):
return abs(cost)
return abs(cost)
info = trade.get("info") or {}
if isinstance(info, dict):
for key in ("fee", "commission", "fillFee"):
try:
v = float(info.get(key) or 0)
if v != 0:
return abs(v)
except (TypeError, ValueError):
continue
return 0.0
def aggregate_bilateral_stats(
fills: list[dict],
*,
contract_size: float = 1.0,
) -> dict[str, float] | None:
"""双边成交额 = 开+平所有相关 fill 的报价币成交额之和;手续费 = fill fee 之和."""
if not fills:
return None
turnover = 0.0
commission = 0.0
for t in fills:
turnover += quote_turnover_usdt_from_fill(t, contract_size=contract_size)
commission += commission_usdt_from_fill(t)
if turnover <= 0 and commission <= 0:
return None
return {
"exchange_turnover_usdt": round(turnover, 4),
"exchange_commission_usdt": round(commission, 4),
}
def filter_position_lifecycle_fills(
trades: list[dict],
direction: str,
open_ms: int | None,
close_ms: int | None,
*,
hedge_mode: bool = False,
close_buffer_ms: int = 15 * 60 * 1000,
) -> list[dict]:
"""
持仓生命周期内 fill:多=开买+平卖;空=开卖+平买.
hedge_mode 时按 posSide 与 direction 过滤.
"""
direction = (direction or "long").strip().lower()
open_side = "buy" if direction == "long" else "sell"
close_side = "sell" if direction == "long" else "buy"
allowed_sides = {open_side, close_side}
upper = int(close_ms) + int(close_buffer_ms) if close_ms else None
out: list[dict] = []
for t in trades or []:
side = (t.get("side") or "").lower()
if side not in allowed_sides:
continue
ts = _coerce_ts_ms(t.get("timestamp"))
if ts is None:
continue
if open_ms and ts < int(open_ms) - 60_000:
continue
if upper and ts > upper:
continue
if hedge_mode:
info = t.get("info") or {}
if not isinstance(info, dict):
info = {}
pos_side = (info.get("posSide") or t.get("posSide") or "").lower()
if pos_side in ("long", "short") and pos_side != direction:
continue
out.append(t)
out.sort(key=lambda x: x.get("timestamp") or 0)
return out
def sum_binance_commission_income(entries: list[dict], trade_ids: set[str] | None) -> float | None:
"""Binance income 流水中 COMMISSION 合计(负值取绝对值为成本)."""
if not entries:
return None
total = 0.0
found = False
for e in entries:
it = (e.get("incomeType") or e.get("income_type") or "").strip()
if it != "COMMISSION":
continue
if trade_ids:
tid = str(e.get("tradeId") or e.get("trade_id") or "").strip()
if tid and tid not in trade_ids:
continue
try:
total += float(e.get("income") or 0)
found = True
except (TypeError, ValueError):
continue
if not found:
return None
return round(abs(total), 4)
def trade_ids_from_fills(fills: list[dict]) -> set[str]:
out: set[str] = set()
for t in fills or []:
info = t.get("info") or {}
if not isinstance(info, dict):
info = {}
for key in ("id", "tradeId", "trade_id"):
raw = t.get(key) if key in t else info.get(key)
if raw is not None and str(raw).strip():
out.add(str(raw).strip())
break
return out
def merge_commission_prefer_income(
fill_commission: float,
income_commission: float | None,
) -> float:
if income_commission is not None and income_commission > 0:
return round(income_commission, 4)
return round(max(fill_commission, 0.0), 4)
def update_trade_record_stats_columns(
conn: Any,
trade_id: int,
turnover_usdt: float | None,
commission_usdt: float | None,
) -> None:
if turnover_usdt is None and commission_usdt is None:
return
conn.execute(
"""
UPDATE trade_records
SET exchange_turnover_usdt = COALESCE(?, exchange_turnover_usdt),
exchange_commission_usdt = COALESCE(?, exchange_commission_usdt)
WHERE id = ?
""",
(turnover_usdt, commission_usdt, int(trade_id)),
)
def attach_exchange_stats_to_trade(
conn: Any,
trade_id: int,
*,
fetch_fills: Callable[[], list[dict]],
contract_size: float = 1.0,
income_commission: float | None = None,
) -> dict[str, float] | None:
"""拉 fill 并写库;仅在新单平仓路径调用."""
try:
fills = fetch_fills() or []
except Exception:
fills = []
stats = aggregate_bilateral_stats(fills, contract_size=contract_size)
if not stats and income_commission is None:
return None
turnover = stats.get("exchange_turnover_usdt") if stats else None
fill_comm = float(stats.get("exchange_commission_usdt") or 0) if stats else 0.0
commission = merge_commission_prefer_income(fill_comm, income_commission)
update_trade_record_stats_columns(
conn,
trade_id,
turnover,
commission if commission > 0 else None,
)
out = {}
if turnover is not None:
out["exchange_turnover_usdt"] = turnover
if commission > 0:
out["exchange_commission_usdt"] = commission
return out or None
lib/trade/trade_fee_lib.py
-94
View File
@@ -1,94 +0,0 @@
"""永续估算盈亏:固定 taker 手续费(默认单边 0.05%,开+平双边).
浮盈亏仍读交易所;本模块只服务「盈利金额 / 止盈盈利 / 推送 / 记账 pnl_amount」等估算口径.
"""
from __future__ import annotations
import math
import os
from typing import Optional
def _finite(v) -> Optional[float]:
try:
f = float(v)
return f if math.isfinite(f) else None
except (TypeError, ValueError):
return None
def taker_fee_rate() -> float:
"""单边 taker 费率,默认 0.0005(=0.05%)."""
raw = os.getenv("PERP_TAKER_FEE_RATE", "0.0005")
rate = _finite(raw)
if rate is None or rate < 0:
return 0.0005
return rate
def notional_usdt(price, qty, contract_size: float = 1.0) -> Optional[float]:
"""名义价值 U = 价格 × 张数 × 合约面值."""
p = _finite(price)
q = _finite(qty)
cs = _finite(contract_size)
if p is None or q is None or p <= 0 or q <= 0:
return None
if cs is None or cs <= 0:
cs = 1.0
return abs(q) * p * cs
def estimate_roundtrip_fee_usdt(
entry_price,
exit_price,
qty=None,
contract_size: float = 1.0,
*,
open_notional: float | None = None,
rate: float | None = None,
) -> float:
"""开+平双边手续费(各单边 rate).
优先用 价×张×面值;若无张数则用 open_notional 估开仓名义,
平仓名义按 exit/entry 缩放.
"""
fee_rate = taker_fee_rate() if rate is None else float(rate)
if fee_rate <= 0:
return 0.0
entry = _finite(entry_price)
exit_p = _finite(exit_price)
open_n = notional_usdt(entry, qty, contract_size) if qty is not None else None
if open_n is None:
open_n = _finite(open_notional)
if open_n is None or open_n <= 0:
return 0.0
if entry is not None and entry > 0 and exit_p is not None and exit_p > 0:
close_n = open_n * (exit_p / entry)
else:
close_n = open_n
return round(open_n * fee_rate + close_n * fee_rate, 8)
def net_pnl_after_fee(
gross_pnl,
entry_price,
exit_price,
qty=None,
contract_size: float = 1.0,
*,
open_notional: float | None = None,
rate: float | None = None,
) -> Optional[float]:
"""毛利扣双边手续费后的净盈亏;gross 无效则返回 None."""
gross = _finite(gross_pnl)
if gross is None:
return None
fee = estimate_roundtrip_fee_usdt(
entry_price,
exit_price,
qty,
contract_size,
open_notional=open_notional,
rate=rate,
)
return round(gross - fee, 4)
lib/trade/trade_labels_lib.py
+2 -9
View File
@@ -158,12 +158,5 @@ def order_monitor_excluded_from_position_limit(conn, row) -> bool:
def count_position_limit_active_monitors(conn) -> int:
try:
rows = conn.execute("SELECT * FROM order_monitors WHERE status='active'").fetchall()
except Exception:
return 0
n = 0
for row in rows:
if not order_monitor_excluded_from_position_limit(conn, row):
n += 1
return n
del conn
return 0
lib/trade/trade_result_lib.py
+13 -1
View File
@@ -2,11 +2,23 @@
from __future__ import annotations
from typing import Optional
from typing import Any, Optional
_WIN_EPS = 1e-9
def snapshot_stop_loss(initial_stop_loss: Any, stop_loss: Any) -> Optional[float]:
"""展示盈亏比时优先用开仓止损快照."""
for v in (initial_stop_loss, stop_loss):
try:
f = float(v)
if f > 0:
return f
except (TypeError, ValueError):
continue
return None
def classify_exit_by_levels(
direction,
trigger_price,
lib/trade/trade_stats_calendar_lib.py
-115
View File
@@ -1,115 +0,0 @@
"""按交易日聚合实例 trade_records 盈亏,供统计分析页日历 API 使用."""
from __future__ import annotations
import json
from datetime import datetime, timedelta
from typing import Any, Callable
def build_trade_stats_calendar(
pnls: list[tuple],
year: int,
month: int,
segment_key: str,
row_matches_fn: Callable[[Any, str], bool],
*,
reset_hour: int = 8,
) -> dict[str, Any]:
"""pnls: _load_completed_trade_pnls 返回值 (pnl, close_dt, trading_day, row)."""
y = int(year)
m = int(month)
if m < 1 or m > 12:
raise ValueError("month 无效")
first = f"{y:04d}-{m:02d}-01"
if m == 12:
next_first = datetime(y + 1, 1, 1)
else:
next_first = datetime(y, m + 1, 1)
last = (next_first - timedelta(days=1)).strftime("%Y-%m-%d")
seg = (segment_key or "all").strip() or "all"
days: dict[str, dict[str, Any]] = {}
for pnl, _close_dt, td, row in pnls:
if not td or td < first or td > last:
continue
if not row_matches_fn(row, seg):
continue
bucket = days.setdefault(
td,
{
"trading_day": td,
"open_count": 0,
"pnl_total": 0.0,
"turnover_total": 0.0,
"commission_total": 0.0,
"has_sick": False,
"sick_count": 0,
},
)
bucket["open_count"] += 1
bucket["pnl_total"] += float(pnl or 0)
try:
bucket["turnover_total"] += float(row["exchange_turnover_usdt"] or 0)
except (TypeError, ValueError, KeyError):
pass
try:
bucket["commission_total"] += float(row["exchange_commission_usdt"] or 0)
except (TypeError, ValueError, KeyError):
pass
for d in days.values():
d["pnl_total"] = round(float(d["pnl_total"]), 4)
d["turnover_total"] = round(float(d["turnover_total"]), 4)
d["commission_total"] = round(float(d["commission_total"]), 4)
month_pnl = sum(float(d["pnl_total"]) for d in days.values())
month_count = sum(int(d["open_count"]) for d in days.values())
return {
"year": y,
"month": m,
"date_from": first,
"date_to": last,
"segment": seg,
"reset_hour": int(reset_hour),
"days": days,
"month_pnl_total": round(month_pnl, 4),
"month_open_count": month_count,
}
def build_initial_stats_calendar(
pnls: list[tuple],
now_dt: datetime,
row_matches_fn: Callable[[Any, str], bool],
*,
reset_hour: int = 8,
segment_key: str = "all",
) -> dict[str, Any]:
"""统计页首屏内嵌日历(当前自然月,默认品类)."""
return build_trade_stats_calendar(
pnls,
now_dt.year,
now_dt.month,
segment_key,
row_matches_fn,
reset_hour=reset_hour,
)
def build_stats_calendar_bootstrap(
pnls: list[tuple],
now_dt: datetime,
row_matches_fn: Callable[[Any, str], bool],
*,
reset_hour: int = 8,
segment_key: str = "all",
) -> tuple[dict[str, Any] | None, str | None]:
"""返回 (payload, json_str);失败时 (None, None),供模板安全内嵌."""
try:
payload = build_initial_stats_calendar(
pnls,
now_dt,
row_matches_fn,
reset_hour=reset_hour,
segment_key=segment_key,
)
return payload, json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
except Exception:
return None, None