fix bug
This commit is contained in:
@@ -17,7 +17,6 @@ from libs.signal import init_signals, SignalItem
|
||||
from libs.collector import collector_push
|
||||
from sdk import Client
|
||||
from libs.grid_take_profit import GridTrailingTracker
|
||||
from .state import State
|
||||
from .order import OrderBook
|
||||
from .watch import DipWatch
|
||||
from .runtime import Runtime
|
||||
@@ -56,14 +55,8 @@ def StartTrend() -> None:
|
||||
portfolio = client.portfolio()
|
||||
assets = portfolio.assets
|
||||
positions = list(portfolio.positions.values())
|
||||
storeState = State.for_strategy(
|
||||
config.global_config.qmt_data_dir,
|
||||
config.account_config.strategy,
|
||||
config.account_config.account_id,
|
||||
)
|
||||
order_book = OrderBook()
|
||||
order_book.refresh(client, portfolio.orders)
|
||||
storeState.reconcile(positions, portfolio.orders)
|
||||
|
||||
# 获取本策略的信号开仓数据
|
||||
signals = init_signals(
|
||||
@@ -76,7 +69,6 @@ def StartTrend() -> None:
|
||||
client=client,
|
||||
global_cfg=config.global_config,
|
||||
account_cfg=config.account_config,
|
||||
state=storeState,
|
||||
orders=order_book,
|
||||
open_watch=DipWatch(),
|
||||
add_watch=DipWatch(),
|
||||
@@ -177,13 +169,6 @@ def RunOnce(run: Runtime, signals:list[SignalItem]) -> None:
|
||||
log.exception("[行情] 获取行情失败，代码数量=%d", len(all_codes))
|
||||
return
|
||||
|
||||
# 6. 更新状态机
|
||||
try:
|
||||
run.state.reconcile(positions, portfolio.orders)
|
||||
except Exception:
|
||||
log.exception("[状态] 订单状态对账失败")
|
||||
return
|
||||
|
||||
log.info("[RunOnce] 本轮就绪，持仓=%d，候选=%d，大盘允许=%s，资金允许=%s", len(positions), len(allow_open), market_ok, allow_open_by_cash)
|
||||
|
||||
# 启动线程，开始计算
|
||||
|
||||
@@ -8,7 +8,6 @@ from libs import calc_buy_volume
|
||||
from sdk import OP_BUY
|
||||
from .runtime import Runtime
|
||||
from .order import PlaceOrderRequest
|
||||
from .state import PendingOrder
|
||||
import logging as log
|
||||
|
||||
|
||||
|
||||
@@ -10,10 +10,9 @@ from sdk import OP_BUY, OP_SELL, PositionItem, Tick
|
||||
|
||||
from .order import PlaceOrderRequest
|
||||
from .runtime import Runtime
|
||||
from .state import PendingOrder
|
||||
import logging as log
|
||||
|
||||
LOSS_TIERS = (-30.0, -50.0)
|
||||
LOSS_TIERS = [-50.0]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
@@ -133,6 +132,8 @@ def handle_profit(
|
||||
)
|
||||
if not runtime.orders.place(runtime.client, request):
|
||||
return TradeDecision(False, "止盈委托失败")
|
||||
|
||||
|
||||
return TradeDecision(True, f"卖出 {volume} 股，订单={order_id}")
|
||||
|
||||
|
||||
@@ -141,19 +142,15 @@ def handle_loss(
|
||||
position: PositionItem,
|
||||
tick: Tick,
|
||||
pnl_rate: float,
|
||||
available: float,
|
||||
available: float
|
||||
) -> TradeDecision:
|
||||
"""按亏损档位、反弹确认和本轮剩余预算提交补仓。"""
|
||||
try:
|
||||
state = runtime.state.get(position.stock_code)
|
||||
except KeyError:
|
||||
return TradeDecision(False, "缺少持仓状态，跳过补仓")
|
||||
|
||||
if state.added_num >= len(LOSS_TIERS):
|
||||
return TradeDecision(False, "已达到最大补仓次数")
|
||||
if pnl_rate > LOSS_TIERS[state.added_num]:
|
||||
add_num = get_add_num(hands=int(position.volume/100),market_value=position.market_value)
|
||||
if add_num >= len(LOSS_TIERS) or add_num < 0:
|
||||
return TradeDecision(False, f"补仓次数无效:{add_num}")
|
||||
if pnl_rate > LOSS_TIERS[add_num]:
|
||||
return TradeDecision(False)
|
||||
if tick.last_price > 200 or position.market_value >= 60_000:
|
||||
if tick.last_price > 200 or position.market_value >= 20_000:
|
||||
return TradeDecision(False, "价格或仓位市值超过补仓限制")
|
||||
if not runtime.add_watch.triggered("补仓", position.stock_code, tick.last_price):
|
||||
return TradeDecision(False, "等待价格反弹确认")
|
||||
@@ -174,15 +171,20 @@ def handle_loss(
|
||||
strategy_name=runtime.account_cfg.strategy,
|
||||
kind="add",
|
||||
)
|
||||
|
||||
#runtime.state.new_order(PendingOrder(order_id, position.stock_code, "added", volume))
|
||||
|
||||
if not runtime.orders.place(runtime.client, request):
|
||||
reserved = amount if runtime.state.busy(position.stock_code) else 0.0
|
||||
return TradeDecision(False, "补仓委托失败或待确认", reserved)
|
||||
return TradeDecision(False, "补仓订单委托失败")
|
||||
|
||||
runtime.add_watch.forget(position.stock_code)
|
||||
return TradeDecision(True, f"买入 {volume} 股，订单={order_id}", amount)
|
||||
|
||||
|
||||
def _position_key(runtime: Runtime, code: str) -> str:
|
||||
return f"{runtime.account_cfg.account_id}:{code}"
|
||||
|
||||
def get_add_num(hands:int,market_value:float) -> int:
|
||||
if market_value>10000:
|
||||
return -1
|
||||
if hands < 2:
|
||||
return 0
|
||||
return -1
|
||||
@@ -10,7 +10,6 @@ from sdk import Client
|
||||
from libs.grid_take_profit import GridTrailingTracker
|
||||
|
||||
from .order import OrderBook
|
||||
from .state import State
|
||||
from .watch import DipWatch
|
||||
|
||||
|
||||
@@ -38,7 +37,6 @@ class Runtime:
|
||||
account_cfg: AccountConfig
|
||||
|
||||
# 策略运行过程中共享的状态组件。
|
||||
state: State
|
||||
orders: OrderBook
|
||||
open_watch: DipWatch
|
||||
add_watch: DipWatch
|
||||
|
||||
@@ -1,190 +0,0 @@
|
||||
"""底仓、补仓记录与待确认订单的 JSON 存储。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging as log
|
||||
import time
|
||||
from dataclasses import asdict, dataclass, field, replace
|
||||
from pathlib import Path
|
||||
from threading import RLock
|
||||
from typing import Any
|
||||
|
||||
from sdk import OrderItem, PositionItem
|
||||
|
||||
PENDING_TIME_OUT = 3600
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class StateItem:
|
||||
code: str
|
||||
base_order_id: str = ""
|
||||
base_qty: int = 0
|
||||
base_cost: float = 0.0
|
||||
added_order_id: str = ""
|
||||
added_num: int = 0
|
||||
added_qty: int = 0
|
||||
added_cost: float = 0.0
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class PendingOrder:
|
||||
order_id: str
|
||||
code: str
|
||||
kind: str
|
||||
pre_qty: int
|
||||
submit_at: int = field(default_factory=lambda: int(time.time()))
|
||||
|
||||
|
||||
class State:
|
||||
def __init__(self, data_dir: str | Path, strategy: str, account_id: str) -> None:
|
||||
self.path = Path(data_dir) / f"{strategy}_{account_id}_state.json"
|
||||
self.lock = RLock()
|
||||
self.items: dict[str, StateItem] = {}
|
||||
self.pending: list[PendingOrder] = []
|
||||
self.IsModify = False
|
||||
self._load()
|
||||
|
||||
def _load(self) -> None:
|
||||
"""启动时读取 JSON 中的持仓记录和待确认订单。"""
|
||||
if self.path.is_file():
|
||||
raw = json.loads(self.path.read_text(encoding="utf-8"))
|
||||
self.items = {code: StateItem(**item) for code, item in raw["items"].items()}
|
||||
self.pending = [PendingOrder(**item) for item in raw["pending"]]
|
||||
now = int(time.time())
|
||||
for pending in list(self.pending):
|
||||
if now - pending.submit_at >= PENDING_TIME_OUT:
|
||||
self.pending.remove(pending)
|
||||
self.IsModify = True
|
||||
log.info("[状态] 清理超时 pending，代码=%s，订单=%s", pending.code, pending.order_id)
|
||||
self.save()
|
||||
|
||||
|
||||
def get(self, code: str) -> StateItem:
|
||||
with self.lock:
|
||||
return self.items[code]
|
||||
|
||||
def new_order(self, order: PendingOrder) -> None:
|
||||
"""提交前保存待确认订单，同一证券已有 pending 时跳过。"""
|
||||
with self.lock:
|
||||
if order:
|
||||
self.pending.append(order)
|
||||
self.IsModify = True
|
||||
self.save()
|
||||
|
||||
def merged_order(self, orders: list[OrderItem]) -> dict[str, dict[str, Any]]:
|
||||
"""按证券代码、本地订单号合并，返回成交数量、金额、均价和状态。"""
|
||||
merged: dict[str, dict[str, Any]] = {}
|
||||
for order in orders:
|
||||
if not order.local_order_id:
|
||||
continue
|
||||
|
||||
if not order.local_order_id in merged.keys():
|
||||
strStatus = "ING"
|
||||
if order.status == "56":
|
||||
strStatus = "OK"
|
||||
merged[order.local_order_id] = {
|
||||
"code":order.code,
|
||||
"qty": order.traded_volume,
|
||||
"cost": order.trade_price,
|
||||
"status": order.status,
|
||||
"merged_status":strStatus
|
||||
}
|
||||
continue
|
||||
|
||||
old = merged[order.local_order_id]
|
||||
strStatus = "ING"
|
||||
if old["status"] == order.status == "56":
|
||||
strStatus = "OK"
|
||||
|
||||
totalQty = old["qty"]+order.traded_volume
|
||||
# 合计成交数量为零，跳过，避免除零。
|
||||
if totalQty == 0:
|
||||
continue
|
||||
# 有成交数量但缺少成交金额，跳过，避免拉低成本。
|
||||
if order.traded_volume > 0 and not order.trade_amount:
|
||||
continue
|
||||
cost = ((old["qty"]*old["cost"])+order.trade_amount) / totalQty
|
||||
merged[order.local_order_id]["qty"] = totalQty
|
||||
merged[order.local_order_id]["cost"] = cost
|
||||
merged[order.local_order_id]["merged_status"] = strStatus
|
||||
|
||||
return merged
|
||||
|
||||
def reconcile(self, positions: list[PositionItem], orders: list[OrderItem]) -> None:
|
||||
"""合并订单 → 对齐 pending 和持仓 → 保存 JSON。"""
|
||||
# 1. 合并同一本地订单的成交数据。
|
||||
merged = self.merged_order(orders)
|
||||
|
||||
with self.lock:
|
||||
# 2. 按本地订单号查找合并结果，核对证券代码后写入底仓或补仓。
|
||||
# 只留下尚未完成确认的订单，避免循环中反复查找、删除列表元素。
|
||||
remaining: list[PendingOrder] = []
|
||||
for pending in self.pending:
|
||||
code = pending.code
|
||||
result = merged.get(pending.order_id)
|
||||
if result is None or result["code"] != code:
|
||||
# 按现有规则：本轮快照中没有对应订单，就清理 pending。
|
||||
self.IsModify = True
|
||||
log.info("[状态] 清理无对应订单的 pending，代码=%s，订单=%s", code, pending.order_id)
|
||||
continue
|
||||
if result["merged_status"] != "OK" or result["qty"] != pending.pre_qty:
|
||||
# 未全部成功，或拆单快照的数量尚未齐全，留到下轮确认。
|
||||
remaining.append(pending)
|
||||
continue
|
||||
qty, cost = result["qty"], result["cost"]
|
||||
# 只记录实际成交。
|
||||
if qty:
|
||||
item = self.items.get(code, StateItem(code))
|
||||
if pending.kind == "base":
|
||||
# 底仓：记录本次成交数量和实际均价。
|
||||
item = replace(item, base_order_id=pending.order_id,
|
||||
base_qty=qty, base_cost=cost)
|
||||
else:
|
||||
# 补仓：次数加一，数量累加，成本按成交数量加权。
|
||||
total_qty = item.added_qty + qty
|
||||
total_amount = item.added_qty * item.added_cost + qty * cost
|
||||
item = replace(item, added_order_id=pending.order_id,
|
||||
added_num=item.added_num + 1,
|
||||
added_qty=total_qty, added_cost=total_amount / total_qty)
|
||||
self.items[code] = item
|
||||
self.IsModify = True
|
||||
log.info("[状态] 对账结束，代码=%s，订单=%s，计划=%d，成交=%d",
|
||||
code, pending.order_id, pending.pre_qty, qty)
|
||||
|
||||
self.pending = remaining
|
||||
|
||||
# 未记录且没有待确认订单的持仓，作为首次接管的底仓导入。
|
||||
pending_codes = {pending.code for pending in remaining}
|
||||
for position in positions:
|
||||
code = position.stock_code
|
||||
if not code or position.volume <= 0:
|
||||
continue
|
||||
item = self.items.get(code)
|
||||
if item is None:
|
||||
if code not in pending_codes:
|
||||
self.items[code] = StateItem(code=code, base_order_id=position.trade_id,
|
||||
base_qty=position.volume, base_cost=position.open_price)
|
||||
self.IsModify = True
|
||||
continue
|
||||
|
||||
# 已有记录只报告数量差异，不用总持仓覆盖底仓/补仓的划分。
|
||||
recorded_qty = item.base_qty + item.added_qty
|
||||
if recorded_qty != position.volume:
|
||||
log.info("[状态] 持仓数量差异，代码=%s，记录=%d，实际=%d",
|
||||
code, recorded_qty, position.volume)
|
||||
|
||||
# 3. 本轮统一保存；没有修改时 save() 不写文件。
|
||||
self.save()
|
||||
|
||||
def save(self) -> None:
|
||||
with self.lock:
|
||||
if not self.IsModify:
|
||||
return
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = self.path.with_suffix(self.path.suffix + ".tmp")
|
||||
payload = dict(items={code: asdict(item) for code, item in self.items.items()},
|
||||
pending=[asdict(item) for item in self.pending])
|
||||
temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False) + "\n", encoding="utf-8")
|
||||
temporary.replace(self.path)
|
||||
self.IsModify = False
|
||||
Reference in New Issue
Block a user