dekun/qihuo
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

feat: 持仓保证金占比与止盈止损自动委托守护

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
2026-06-24 14:20:15 +08:00
parent 23d0f1d6fa
commit 73b9dfdfdb
8 changed files with 751 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
install_trading.py
+262 -41
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import json
import logging
from datetime import datetime
from typing import Any, Callable, Optional
@@ -18,10 +19,17 @@ from position_sizing import (
calc_order_tick_metrics,
normalize_sizing_mode,
)
from recommend_store import load_recommend_cache, refresh_recommend_cache
from recommend_store import load_recommend_cache, recommend_payload, refresh_recommend_cache
from recommend_stream import recommend_hub, start_recommend_worker
from ctp_reconnect import start_ctp_reconnect_worker
from ctp_fee_worker import start_ctp_fee_worker
from sl_tp_guard import (
ensure_monitor_order_columns,
monitor_order_status,
place_monitor_exit_orders,
start_sl_tp_guard_worker,
sync_all_sl_tp_orders,
)
from risk.account_risk_lib import (
assert_can_open,
get_risk_status,
@@ -58,6 +66,9 @@ from vnpy_bridge import (
)
logger = logging.getLogger(__name__)
def install_trading(app, *, login_required, require_nav, get_db, get_setting, set_setting, fetch_price, send_wechat_msg):
"""注册交易相关路由。"""
_nav = require_nav
@@ -238,10 +249,13 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
mode = get_trading_mode(get_setting)
ctp_st = ctp_status(mode)
rows: list[dict] = []
capital = _capital(conn)
if not ctp_st.get("connected"):
return rows
ensure_monitor_order_columns(conn)
# 程序监控仅用于补充止损/止盈，持仓以 CTP 柜台为准
monitor_map: dict[tuple[str, str], dict] = {}
for r in conn.execute(
@@ -271,6 +285,46 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
break
sl = float(mon["stop_loss"]) if mon and mon.get("stop_loss") is not None else None
tp = float(mon["take_profit"]) if mon and mon.get("take_profit") is not None else None
open_time = (mon.get("open_time") or "") if mon else ""
holding = _holding_duration(open_time, now_iso) if open_time else ""
mark = None
if codes:
mark = fetch_price(
sym,
codes.get("market_code", ""),
codes.get("sina_code", ""),
)
close_est = float(mark) if mark and mark > 0 else entry
fee_info = calc_fee_breakdown(
sym,
entry,
close_est,
lots,
open_time or now_iso,
now_iso,
trading_mode=mode,
)
est_net = None
if float_pnl is not None:
est_net = round(float(float_pnl) - fee_info["total_fee"], 2)
pos_metrics = calc_position_metrics(
direction,
entry,
sl if sl is not None else entry,
tp if tp is not None else entry,
lots,
mark,
capital,
sym,
)
order_st = monitor_order_status(
mon or {}, mode=mode, ths_code=sym, direction=direction,
)
can_place = bool(
mon
and (mon.get("stop_loss") is not None or mon.get("take_profit") is not None)
and (order_st.get("needs_sl_order") or order_st.get("needs_tp_order"))
)
pending_for_row: list[dict] = []
if sl is not None:
pending_for_row.append({
@@ -303,8 +357,21 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
"entry_price": entry,
"stop_loss": sl,
"take_profit": tp,
"mark_price": None,
"open_time": open_time or None,
"holding_duration": holding or None,
"mark_price": mark,
"current_price": mark,
"margin": pos_metrics.get("margin"),
"position_pct": pos_metrics.get("position_pct"),
"float_pnl": float_pnl,
"est_fee": fee_info["total_fee"],
"est_fee_open": fee_info["open_fee"],
"est_fee_close": fee_info["close_fee"],
"est_fee_close_type": fee_info["close_type"],
"est_pnl_net": est_net,
"sl_order_active": order_st.get("sl_order_active"),
"tp_order_active": order_st.get("tp_order_active"),
"can_place_orders": can_place,
"tick_value_total": tick.get("tick_value_total"),
"price_precision": tick.get("price_precision"),
"tick_size": tick.get("tick_size"),
@@ -341,7 +408,7 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
).fetchone()["n"]
conn.commit()
sizing = get_sizing_mode(get_setting)
rec_cache = load_recommend_cache(conn)
rec_cache = recommend_payload(conn, live_capital=capital)
return render_template(
"trade.html",
trading_mode=mode,
@@ -376,6 +443,7 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
mode = get_trading_mode(get_setting)
ctp_st = ctp_status(mode)
_sync_trade_monitors_with_ctp(conn, mode)
sync_all_sl_tp_orders(conn, mode)
rows = _build_trading_live_rows(conn)
pending_orders = _build_pending_orders(conn, mode)
capital = _capital(conn)
@@ -392,6 +460,143 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
finally:
conn.close()
@app.route("/api/trading/monitor/upsert", methods=["POST"])
@login_required
def api_trading_monitor_upsert():
"""为已有 CTP 持仓补充/更新本地止盈止损监控。"""
d = request.get_json(silent=True) or {}
sym = (d.get("symbol_code") or d.get("symbol") or "").strip()
direction = (d.get("direction") or "long").strip().lower()
try:
lots = max(1, int(d.get("lots") or 1))
entry = float(d.get("entry_price") or d.get("entry") or 0)
sl = float(d["stop_loss"]) if d.get("stop_loss") not in (None, "") else None
tp = float(d["take_profit"]) if d.get("take_profit") not in (None, "") else None
except (TypeError, ValueError, KeyError):
return jsonify({"ok": False, "error": "参数无效"}), 400
if not sym:
return jsonify({"ok": False, "error": "缺少品种代码"}), 400
if sl is None and tp is None:
return jsonify({"ok": False, "error": "请至少填写止损或止盈"}), 400
mode = get_trading_mode(get_setting)
if not ctp_status(mode).get("connected"):
return jsonify({"ok": False, "error": "请先连接 CTP"}), 400
has_pos = False
for p in _ctp_positions(mode):
if int(p.get("lots") or 0) <= 0:
continue
if (p.get("direction") or "long") != direction:
continue
if _match_ctp_symbol(p.get("symbol") or "", sym):
has_pos = True
lots = int(p.get("lots") or lots)
entry = float(p.get("avg_price") or entry or 0)
sym = (p.get("symbol") or sym).strip()
break
if not has_pos:
return jsonify({"ok": False, "error": "柜台无对应持仓"}), 400
conn = get_db()
try:
init_strategy_tables(conn)
mon = None
for r in conn.execute(
"SELECT * FROM trade_order_monitors WHERE status='active'"
).fetchall():
row = dict(r)
if row.get("direction") != direction:
continue
if _match_ctp_symbol(sym, row.get("symbol") or ""):
mon = row
break
codes = ths_to_codes(sym)
now_s = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if mon:
conn.execute(
"""UPDATE trade_order_monitors SET stop_loss=?, take_profit=?, lots=?, entry_price=?
WHERE id=?""",
(sl, tp, lots, entry or mon.get("entry_price"), mon["id"]),
)
mid = mon["id"]
else:
conn.execute(
"""INSERT INTO trade_order_monitors (
symbol, symbol_name, market_code, direction, lots, entry_price,
stop_loss, take_profit, open_time, monitor_type, status
) VALUES (?,?,?,?,?,?,?,?,?,?, 'active')""",
(
sym,
codes.get("name", sym) if codes else sym,
codes.get("market_code", "") if codes else "",
direction,
lots,
entry,
sl,
tp,
now_s,
"manual",
),
)
mid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
conn.commit()
mon_row = conn.execute(
"SELECT * FROM trade_order_monitors WHERE id=?", (mid,),
).fetchone()
if mon_row and (sl is not None or tp is not None):
try:
ensure_monitor_order_columns(conn)
place_monitor_exit_orders(conn, dict(mon_row), mode=mode, force=False)
except Exception as exc:
logger.warning("补充止盈止损后自动委托失败: %s", exc)
return jsonify({"ok": True, "monitor_id": mid, "message": "止盈止损已保存"})
finally:
conn.close()
@app.route("/api/trading/monitor/place-orders", methods=["POST"])
@login_required
def api_trading_monitor_place_orders():
"""按开仓快照向 CTP 挂止盈止损平仓委托。"""
d = request.get_json(silent=True) or {}
try:
monitor_id = int(d.get("monitor_id") or 0)
except (TypeError, ValueError):
monitor_id = 0
conn = get_db()
try:
init_strategy_tables(conn)
ensure_monitor_order_columns(conn)
mode = get_trading_mode(get_setting)
if not ctp_status(mode).get("connected"):
return jsonify({"ok": False, "error": "请先连接 CTP"}), 400
mon = None
if monitor_id > 0:
row = conn.execute(
"SELECT * FROM trade_order_monitors WHERE id=? AND status='active'",
(monitor_id,),
).fetchone()
mon = dict(row) if row else None
if not mon:
sym = (d.get("symbol_code") or "").strip()
direction = (d.get("direction") or "long").strip().lower()
for r in conn.execute(
"SELECT * FROM trade_order_monitors WHERE status='active'"
).fetchall():
row = dict(r)
if row.get("direction") != direction:
continue
if _match_ctp_symbol(sym, row.get("symbol") or ""):
mon = row
break
if not mon:
return jsonify({"ok": False, "error": "未找到有效监控快照"}), 404
result = place_monitor_exit_orders(
conn, mon, mode=mode, force=bool(d.get("force")),
)
if not result.get("ok"):
return jsonify(result), 400
return jsonify(result)
finally:
conn.close()
@app.route("/api/trading/monitor/dismiss", methods=["POST"])
@login_required
def api_trading_monitor_dismiss():
@@ -646,40 +851,49 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
if offset.startswith("open"):
sl = d.get("stop_loss")
tp = d.get("take_profit")
if sl or tp:
import time
time.sleep(2.0)
actual_lots = lots
has_pos = False
for p in _ctp_positions(mode):
if int(p.get("lots") or 0) <= 0:
continue
if (p.get("direction") or "long") != direction:
continue
if _match_ctp_symbol(p.get("symbol") or "", sym):
has_pos = True
actual_lots = int(p.get("lots") or lots)
break
if has_pos:
codes = ths_to_codes(sym)
conn.execute(
"""INSERT INTO trade_order_monitors (
symbol, symbol_name, market_code, direction, lots, entry_price,
stop_loss, take_profit, open_time, monitor_type, status
) VALUES (?,?,?,?,?,?,?,?,?,?, 'active')""",
(
sym,
codes.get("name", sym) if codes else sym,
codes.get("market_code", "") if codes else "",
direction,
actual_lots,
price,
float(sl) if sl else None,
float(tp) if tp else None,
datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"manual",
),
)
import time
time.sleep(2.0)
actual_lots = lots
has_pos = False
for p in _ctp_positions(mode):
if int(p.get("lots") or 0) <= 0:
continue
if (p.get("direction") or "long") != direction:
continue
if _match_ctp_symbol(p.get("symbol") or "", sym):
has_pos = True
actual_lots = int(p.get("lots") or lots)
break
if has_pos:
codes = ths_to_codes(sym)
conn.execute(
"""INSERT INTO trade_order_monitors (
symbol, symbol_name, market_code, direction, lots, entry_price,
stop_loss, take_profit, open_time, monitor_type, status
) VALUES (?,?,?,?,?,?,?,?,?,?, 'active')""",
(
sym,
codes.get("name", sym) if codes else sym,
codes.get("market_code", "") if codes else "",
direction,
actual_lots,
price,
float(sl) if sl else None,
float(tp) if tp else None,
datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"manual",
),
)
mid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
mon_row = conn.execute(
"SELECT * FROM trade_order_monitors WHERE id=?", (mid,),
).fetchone()
if mon_row and (sl or tp):
try:
ensure_monitor_order_columns(conn)
place_monitor_exit_orders(conn, dict(mon_row), mode=mode, force=False)
except Exception as exc:
logger.warning("开仓后自动挂止盈止损失败: %s", exc)
conn.commit()
send_wechat_msg(f"{trading_mode_label(get_setting)} {offset} {sym} {direction} {lots}手 @{price}")
conn.close()
@@ -760,7 +974,7 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
"""只读数据库缓存，不在请求时拉行情。"""
conn = get_db()
try:
payload = load_recommend_cache(conn)
payload = recommend_payload(conn, live_capital=_capital(conn))
return jsonify({"ok": True, **payload})
finally:
conn.close()
@@ -775,7 +989,7 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
try:
conn = get_db()
try:
payload = load_recommend_cache(conn)
payload = recommend_payload(conn, live_capital=_capital(conn))
finally:
conn.close()
yield sse_format("recommend", {"ok": True, **payload})
@@ -806,8 +1020,9 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
try:
init_strategy_tables(conn)
capital = _capital(conn)
rows = refresh_recommend_cache(conn, capital, _main_quote)
payload = load_recommend_cache(conn)
mode = get_trading_mode(get_setting)
rows = refresh_recommend_cache(conn, capital, _main_quote, trading_mode=mode)
payload = recommend_payload(conn, live_capital=capital)
recommend_hub.broadcast("recommend", {"ok": True, **payload})
return jsonify({"ok": True, "count": len(rows), **payload})
finally:
@@ -1139,8 +1354,14 @@ def install_trading(app, *, login_required, require_nav, get_db, get_setting, se
get_capital_fn=_capital,
quote_fn=_main_quote,
init_tables_fn=_init_tables,
get_mode_fn=lambda: get_trading_mode(get_setting),
)
start_ctp_reconnect_worker(get_mode_fn=lambda: get_trading_mode(get_setting))
start_sl_tp_guard_worker(
db_path=DB_PATH,
get_mode_fn=lambda: get_trading_mode(get_setting),
init_tables_fn=_init_tables,
)
start_ctp_fee_worker(
get_mode_fn=lambda: get_trading_mode(get_setting),
get_setting_fn=get_setting,