dekun
6e423eebfb
新增 vnpy CTP 桥接、以损定仓/固定张数、趋势回调与滚仓策略、按资金推荐品种及交易风控；模拟盘走 SimNow，实盘预留期货公司配置。 Co-authored-by: Cursor <cursoragent@cursor.com>
109 lines
3.8 KiB
Python
109 lines
3.8 KiB
Python
"""趋势回调：纯计算（期货整数手）。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
from typing import Any, Optional, Tuple
|
|
|
|
from contract_specs import get_contract_spec
|
|
|
|
|
|
def validate_trend_bounds(direction: str, stop_loss: float, add_upper: float) -> Optional[str]:
|
|
direction = (direction or "long").strip().lower()
|
|
if direction == "long":
|
|
if not (float(stop_loss) < float(add_upper)):
|
|
return "做多：止损须低于补仓上沿"
|
|
else:
|
|
if not (float(stop_loss) > float(add_upper)):
|
|
return "做空：止损须高于补仓下沿"
|
|
return None
|
|
|
|
|
|
def build_grid_prices(direction: str, sl: float, upper: float, n_legs: int) -> list[float]:
|
|
sl, upper = float(sl), float(upper)
|
|
out: list[float] = []
|
|
if n_legs <= 0:
|
|
return out
|
|
direction = (direction or "long").strip().lower()
|
|
if direction == "long":
|
|
if upper <= sl:
|
|
return out
|
|
span = upper - sl
|
|
for i in range(1, n_legs + 1):
|
|
out.append(sl + (i / float(n_legs + 1)) * span)
|
|
out.sort(reverse=True)
|
|
else:
|
|
if sl <= upper:
|
|
return out
|
|
span = sl - upper
|
|
for i in range(1, n_legs + 1):
|
|
out.append(upper + (i / float(n_legs + 1)) * span)
|
|
out.sort()
|
|
return [round(p, 4) for p in out]
|
|
|
|
|
|
def compute_trend_plan_futures(
|
|
*,
|
|
direction: str,
|
|
stop_loss: float,
|
|
add_upper: float,
|
|
take_profit: float,
|
|
risk_percent: float,
|
|
capital: float,
|
|
live_price: float,
|
|
ths_code: str,
|
|
dca_legs: int = 5,
|
|
) -> Tuple[Optional[dict[str, Any]], Optional[str]]:
|
|
err = validate_trend_bounds(direction, stop_loss, add_upper)
|
|
if err:
|
|
return None, err
|
|
spec = get_contract_spec(ths_code)
|
|
mult = spec["mult"]
|
|
d = (direction or "long").strip().lower()
|
|
if d == "short":
|
|
worst_per_lot = (float(stop_loss) - float(add_upper)) * mult
|
|
else:
|
|
worst_per_lot = (float(add_upper) - float(stop_loss)) * mult
|
|
if worst_per_lot <= 0:
|
|
return None, "止损与补仓边界无法计算风险"
|
|
budget = float(capital) * float(risk_percent) / 100.0
|
|
total_lots = int(math.floor(budget / worst_per_lot))
|
|
if total_lots < 3:
|
|
return None, f"按 {risk_percent}% 风险，总手数至少需 3 手才能拆分首仓+补仓（当前 {total_lots} 手）"
|
|
first_lots = total_lots // 2
|
|
remainder = total_lots - first_lots
|
|
legs = max(1, min(int(dca_legs), remainder))
|
|
per_leg = remainder // legs
|
|
leg_amounts = [per_leg] * (legs - 1) + [remainder - per_leg * (legs - 1)]
|
|
if any(x < 1 for x in leg_amounts):
|
|
legs = 1
|
|
leg_amounts = [remainder]
|
|
grid = build_grid_prices(d, stop_loss, add_upper, len(leg_amounts))
|
|
margin_rate = spec["margin_rate"]
|
|
plan_margin = float(live_price) * mult * total_lots * margin_rate
|
|
return {
|
|
"direction": d,
|
|
"stop_loss": float(stop_loss),
|
|
"add_upper": float(add_upper),
|
|
"take_profit": float(take_profit),
|
|
"risk_percent": float(risk_percent),
|
|
"capital_snapshot": float(capital),
|
|
"live_price_ref": float(live_price),
|
|
"target_lots": total_lots,
|
|
"first_lots": first_lots,
|
|
"remainder_lots": remainder,
|
|
"dca_legs": len(leg_amounts),
|
|
"leg_amounts": leg_amounts,
|
|
"leg_amounts_json": json.dumps(leg_amounts),
|
|
"grid_prices_json": json.dumps(grid),
|
|
"grid": grid,
|
|
"plan_margin": round(plan_margin, 2),
|
|
"mult": mult,
|
|
}, None
|
|
|
|
|
|
def trend_dca_level_reached(direction: str, mark_price: float, level: float) -> bool:
|
|
d = (direction or "long").strip().lower()
|
|
pf, lv = float(mark_price), float(level)
|
|
return pf <= lv if d == "long" else pf >= lv
|