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

Fix TP/SL exit classification when exchange fill slips past the tight band.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-08-01 09:38:28 +08:00
parent 4b4dca9e3c
commit a00699aec3
5 changed files with 80 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_monitor_binance/app.py
+1 -23
View File
@@ -257,6 +257,7 @@ from lib.common.history_window_lib import (
utc_window_to_utc_sql_strings,
)
from lib.trade.trade_result_lib import (
classify_exit_by_levels,
count_winning_trades,
filter_trade_records_excluding_miss,
normalize_result_with_pnl,
@@ -4261,29 +4262,6 @@ def ms_to_app_local_str(ms):
return app_now_str()
def classify_exit_by_levels(direction, trigger_price, stop_loss, take_profit, exit_price):
"""根据成交价相对止盈/止损位归类;无法可靠归类时返回 None."""
try:
tp = float(take_profit)
sl = float(stop_loss)
ex = float(exit_price)
trig = float(trigger_price)
except (TypeError, ValueError):
return None
band = max(abs(trig) * 0.0008, abs(tp - sl) * 0.003, 1e-12)
if direction == "long":
if ex >= tp - band:
return "止盈"
if ex <= sl + band:
return "止损"
else:
if ex <= tp + band:
return "止盈"
if ex >= sl - band:
return "止损"
return None
def fetch_latest_closing_fill(exchange_symbol, direction, opened_at_str, opened_at_ms=None):
"""取开仓以来最近一笔减仓成交(与方向一致);失败返回 None."""
if not (BINANCE_API_KEY and BINANCE_API_SECRET):
crypto_monitor_gate/app.py
+1 -23
View File
@@ -260,6 +260,7 @@ from lib.common.history_window_lib import (
utc_window_to_utc_sql_strings,
)
from lib.trade.trade_result_lib import (
classify_exit_by_levels,
count_winning_trades,
filter_trade_records_excluding_miss,
normalize_result_with_pnl,
@@ -3892,29 +3893,6 @@ def ms_to_app_local_str(ms):
return app_now_str()
def classify_exit_by_levels(direction, trigger_price, stop_loss, take_profit, exit_price):
"""根据成交价相对止盈/止损位归类;无法可靠归类时返回 None."""
try:
tp = float(take_profit)
sl = float(stop_loss)
ex = float(exit_price)
trig = float(trigger_price)
except (TypeError, ValueError):
return None
band = max(abs(trig) * 0.0008, abs(tp - sl) * 0.003, 1e-12)
if direction == "long":
if ex >= tp - band:
return "止盈"
if ex <= sl + band:
return "止损"
else:
if ex <= tp + band:
return "止盈"
if ex >= sl - band:
return "止损"
return None
def fetch_latest_closing_fill(exchange_symbol, direction, opened_at_str, opened_at_ms=None):
"""取开仓以来最近一笔减仓成交(与方向一致);失败返回 None."""
if not (GATE_API_KEY and GATE_API_SECRET):
crypto_monitor_okx/app.py
+1 -23
View File
@@ -256,6 +256,7 @@ from lib.common.history_window_lib import (
utc_window_to_utc_sql_strings,
)
from lib.trade.trade_result_lib import (
classify_exit_by_levels,
count_winning_trades,
filter_trade_records_excluding_miss,
normalize_result_with_pnl,
@@ -3381,29 +3382,6 @@ def ms_to_app_local_str(ms):
return app_now_str()
def classify_exit_by_levels(direction, trigger_price, stop_loss, take_profit, exit_price):
"""根据成交价相对止盈/止损位归类;无法可靠归类时返回 None."""
try:
tp = float(take_profit)
sl = float(stop_loss)
ex = float(exit_price)
trig = float(trigger_price)
except (TypeError, ValueError):
return None
band = max(abs(trig) * 0.0008, abs(tp - sl) * 0.003, 1e-12)
if direction == "long":
if ex >= tp - band:
return "止盈"
if ex <= sl + band:
return "止损"
else:
if ex <= tp + band:
return "止盈"
if ex >= sl - band:
return "止损"
return None
def fetch_latest_closing_fill(exchange_symbol, direction, opened_at_str, opened_at_ms=None):
"""取开仓以来最近一笔减仓成交(与方向一致);失败返回 None."""
if not (OKX_API_KEY and OKX_API_SECRET and OKX_API_PASSPHRASE):
lib/trade/trade_result_lib.py
+59
View File
@@ -1,8 +1,67 @@
"""交易结果展示与入库时的语义归一化."""
from __future__ import annotations
from typing import Optional
_WIN_EPS = 1e-9
def classify_exit_by_levels(
direction,
trigger_price,
stop_loss,
take_profit,
exit_price,
) -> Optional[str]:
"""根据成交价相对止盈/止损位归类;无法可靠归类时返回 None.
交易所条件止盈常按标记价触发、市价成交,成交价可能偏离计划止盈数个 tick.
因此先用窄带,失败后再用宽带;仍失败则看是否落在入场→止盈/止损的「盈利/亏损侧」。
"""
try:
tp = float(take_profit)
sl = float(stop_loss)
ex = float(exit_price)
trig = float(trigger_price)
except (TypeError, ValueError):
return None
d = (direction or "").strip().lower()
if d not in ("long", "short"):
return None
band = max(abs(trig) * 0.0008, abs(tp - sl) * 0.003, 1e-12)
# 宽带:覆盖 BTC 等高价币种条件单滑点(实测 Gate 止盈成交可偏出窄带 ~100U)
band_loose = max(abs(trig) * 0.003, abs(tp - sl) * 0.05, band * 4.0, 1e-12)
def _is_tp(b: float) -> bool:
return ex >= tp - b if d == "long" else ex <= tp + b
def _is_sl(b: float) -> bool:
return ex <= sl + b if d == "long" else ex >= sl - b
if _is_tp(band):
return "止盈"
if _is_sl(band):
return "止损"
if _is_tp(band_loose):
return "止盈"
if _is_sl(band_loose):
return "止损"
# 盈利侧且更靠近止盈 → 止盈; 亏损侧且更靠近止损 → 止损
if d == "long":
if ex > trig and abs(ex - tp) <= abs(ex - trig):
return "止盈"
if ex < trig and abs(ex - sl) <= abs(ex - trig):
return "止损"
else:
if ex < trig and abs(ex - tp) <= abs(ex - trig):
return "止盈"
if ex > trig and abs(ex - sl) <= abs(ex - trig):
return "止损"
return None
def normalize_display_result(result):
"""展示用:外部平仓一律视为手动平仓."""
res = (result or "").strip()
tests/test_trade_result_lib.py
+18 -1
View File
@@ -1,4 +1,21 @@
from lib.trade.trade_result_lib import normalize_result_with_pnl, normalize_display_result, is_winning_pnl
from lib.trade.trade_result_lib import (
classify_exit_by_levels,
normalize_result_with_pnl,
normalize_display_result,
is_winning_pnl,
)
def test_classify_short_tp_with_slippage_past_tight_band():
# Gate BTC 空单:计划止盈 62600,实际成交约 62738(条件单滑点),旧窄带会误判为外部平仓
assert (
classify_exit_by_levels("short", 64256.2, 65400, 62600, 62738.0) == "止盈"
)
def test_classify_long_sl_and_tp_basic():
assert classify_exit_by_levels("long", 100, 95, 110, 110.05) == "止盈"
assert classify_exit_by_levels("long", 100, 95, 110, 94.9) == "止损"
def test_stop_loss_with_profit_becomes_trailing_tp():