fix(okx): show negative unrealized PnL on strategy page

Parse signed upl/unrealizedPnl from CCXT positions and fall back to calc_pnl when exchange metrics are missing.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-02 16:02:24 +08:00
parent 724b4a3dd8
commit a5f4ad8e97
3 changed files with 83 additions and 2 deletions
+18 -1
View File
@@ -2789,10 +2789,12 @@ def parse_ccxt_position_metrics(position, order_leverage=None):
initial = approx
except (TypeError, ValueError):
pass
unrealized = _coerce_float(
unrealized = _coerce_float_signed(
p.get("unrealizedPnl"),
info.get("upl"),
info.get("uplLast"),
info.get("unrealized_pnl"),
info.get("unrealisedPnl"),
)
mark = _coerce_float(p.get("markPrice"), p.get("mark_price"), info.get("markPx"))
out = {}
@@ -3996,6 +3998,7 @@ def calc_price_diff_pct(current_price, target_price):
def _coerce_float(*values):
"""取第一个可解析且 > 0 的数(用于价格、保证金等)。"""
for v in values:
if v is None:
continue
@@ -4008,6 +4011,20 @@ def _coerce_float(*values):
return None
def _coerce_float_signed(*values):
"""取第一个有限浮点数(含 0 与负数),用于未实现盈亏等。"""
for v in values:
if v is None or v == "":
continue
try:
f = float(v)
if math.isfinite(f):
return f
except (TypeError, ValueError):
continue
return None
def _sqlite_row_val(row, key, default=None):
try:
v = row[key]
+27 -1
View File
@@ -276,9 +276,35 @@ def enrich_trend_plan(cfg: dict, row) -> dict:
direction = (d.get("direction") or "long").lower()
metrics_fn = getattr(m, "get_live_position_exchange_metrics", None)
if callable(metrics_fn):
met = metrics_fn(ex_sym, direction)
try:
lev = int(d.get("leverage") or 0) or None
except (TypeError, ValueError):
lev = None
try:
met = metrics_fn(ex_sym, direction, order_leverage=lev)
except TypeError:
met = metrics_fn(ex_sym, direction)
if met and met.get("unrealized_pnl") is not None:
d["floating_pnl"] = float(met["unrealized_pnl"])
elif (
met
and met.get("mark_price") is not None
and d.get("avg_entry_price") is not None
):
try:
entry = float(d["avg_entry_price"])
mark = float(met["mark_price"])
margin = float(d.get("plan_margin_capital") or 0)
leverage = int(d.get("leverage") or 1)
calc_pnl = getattr(m, "calc_pnl", None)
if callable(calc_pnl) and entry > 0 and margin > 0:
d["floating_pnl"] = float(
calc_pnl(direction, entry, mark, margin, leverage)
)
else:
d["floating_pnl"] = None
except (TypeError, ValueError):
d["floating_pnl"] = None
else:
d["floating_pnl"] = None
if met and met.get("mark_price") is not None:
+38
View File
@@ -0,0 +1,38 @@
"""OKX 持仓指标解析:未实现盈亏须支持负数。"""
from __future__ import annotations
import unittest
class TestOkxPositionMetrics(unittest.TestCase):
def test_parse_unrealized_pnl_negative(self):
from crypto_monitor_okx.app import parse_ccxt_position_metrics
pos = {
"side": "long",
"contracts": 10,
"markPrice": 0.43,
"unrealizedPnl": -1.25,
"info": {"upl": "-1.25", "markPx": "0.43"},
}
out = parse_ccxt_position_metrics(pos, order_leverage=5)
self.assertIsNotNone(out)
self.assertAlmostEqual(out["unrealized_pnl"], -1.25)
self.assertAlmostEqual(out["mark_price"], 0.43)
def test_parse_unrealized_pnl_zero(self):
from crypto_monitor_okx.app import parse_ccxt_position_metrics
pos = {
"side": "long",
"contracts": 1,
"unrealizedPnl": 0,
"info": {"upl": "0"},
}
out = parse_ccxt_position_metrics(pos)
self.assertIsNotNone(out)
self.assertEqual(out["unrealized_pnl"], 0.0)
if __name__ == "__main__":
unittest.main()