This commit is contained in:
2026-09-11 17:40:51 +08:00
parent af790bca28
commit bcb03e2ed9
4 changed files with 317 additions and 126 deletions

View File

@@ -87,6 +87,10 @@ class State:
db.row_factory = sqlite3.Row
return db
def get_by_code(self,code: str) -> dict:
s = self.state.get(code,{})
return s
def load(self) -> None:
"""缓存状态表和成交记录。"""
self.load_state()

View File

@@ -110,7 +110,7 @@ def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None:
started_at = time.monotonic()
futures: list[tuple[str, Future]] = []
try:
deals = run.client.deals()
portfolio = run.client.portfolio()
@@ -172,7 +172,7 @@ def RunOnce(run: Runtime, state: State, signals: list[SignalItem]) -> None:
(
"持仓计算",
run.executor.submit(
manage_positions, run, ticks, positions, market_ok, assets.available
manage_positions, run, ticks, positions, market_ok, assets.available, state
),
)
)

View File

@@ -1,56 +1,129 @@
"""使用 dcm 信号建立做 T 底仓"""
"""趋势策略开仓逻辑"""
from datetime import datetime
import logging as log
from functools import lru_cache
import math
from libs import calc_buy_volume
from sdk import OP_BUY
from libs.runtime import Runtime
from libs.order import PlaceOrderRequest
import logging as log
def open_signal(run: Runtime, ticks, signals, available: float) -> float:
"""逐个验证开仓信号并提交买入委托,返回本轮剩余资金"""
now = datetime.now()
if (now.hour, now.minute) >= (14, 50):
return available
for item in signals:
def open_signal(run: Runtime, ticks, open_signals) -> None:
"""逐个验证开仓信号并提交买入委托。"""
for item in open_signals:
try:
if not math.isfinite(item.last_close) or item.last_close <= 0:
log.info("[OpenSkip] %s 信号=%s跳过信号无效last_close不是有限正数", item.code, item.signal_key)
continue
if item.code in run.account_cfg.excluded_codes:
log.info("[OpenSkip] %s 信号=%s,跳过:已配置为排除股票", item.code, item.signal_key)
continue
# 由委托簿检查活动委托,防止重复下单。
if (
run.orders.busy(item.code, "BUY")
or run.orders.busy(item.code, "SELL")
):
# 1. 验证信号配置允许开仓的时间区间。
signal_config = run.global_cfg.signals.get(item.signal_key)
if signal_config is None:
log.info("[OpenSkip] %s 信号=%s,跳过:未找到信号配置",item.code,item.signal_key)
continue
# 行情无效或超过策略价格上限时跳过。
if not check_timezone(signal_config.timezone):
log.info("[OpenSkip] %s 信号=%s,跳过:不在信号时间段(%s)",item.code,item.signal_key,signal_config.timezone)
continue
# 2. 检查该证券是否已有买入委托锁,防止重复下单。
if run.orders.busy(item.code, "BUY"):
log.info("[OpenSkip] %s 信号=%s,跳过:买入委托处理中", item.code, item.signal_key)
continue
# 3. 验证行情和最新价格是否有效。
tick = ticks.get(item.code)
price = tick.last_price if tick else 0.0
if (
not math.isfinite(price)
or price <= 0
or price > run.account_cfg.zt_max_price
):
price = tick.last_price if tick is not None else 0
if not math.isfinite(price) or price <= 0:
log.info("[OpenSkip] %s 信号=%s,跳过:价格无效", item.code, item.signal_key)
continue
# 根据单笔买入金额计算整手数量,并预留少量价差和费用。
budget = min(run.account_cfg.buy_value, available)
volume = int(budget / (price * 1.01)) // 100 * 100
amount = price * volume * 1.01
if volume < (200 if item.code.startswith('688') else 100):
# 5. 根据单笔买入金额计算整手开仓数量。
volume = calc_buy_volume(price, run.account_cfg.buy_value)
if volume <= 0:
log.info("[OpenSkip] %s 信号=%s,跳过:数量无效", item.code, item.signal_key)
continue
# 等待价格从观察低点反弹,防止直接接下跌中的“飞刀”。
if not run.open_watch.triggered("ZT 建仓", item.code, price):
# 其它信号,均从观察低点反弹,防止直接接下跌中的“飞刀”。
if not run.open_watch.triggered("开仓", item.code, price):
continue
order_id = run.orders.new_order_id("zt", "base")
request = PlaceOrderRequest(
OP_BUY, item.code, volume, order_id, "zt"
)
# 即使响应丢失,本轮也预留资金;状态簿只在取得实际成交后入账。
available -= amount
if run.orders.place(run.client, request):
run.open_watch.forget(item.code)
log.info("[ZT 建仓] %s 买入 %d 股,等待实际成交", item.code, volume)
except Exception:
log.exception("[ZT 建仓] %s 处理异常,继续后续信号", item.code)
return available
do_open(run, item.code, volume, item.signal_key, price)
except RuntimeError as exc:
log.exception("[OpenRuntimeError] %s 信号=%s,失败:%s",item.code,item.signal_key,exc)
except Exception as err:
log.exception("[OpenExceptionError] %s 信号=%s,异常:%s",item.code,item.signal_key,err)
continue
def do_open(
run: Runtime, code: str, volume: int, signal_key: str, price: float
) -> None:
"""生成本地订单号并按最新价提交开仓委托。"""
order_id = run.orders.new_order_id("zt","base")
request = PlaceOrderRequest(
OP_BUY,
code,
volume,
order_id,
signal_key,
kind="base",
)
if not run.orders.place(run.client, request):
raise RuntimeError("订单提交失败")
run.open_watch.forget(code)
log.info("[Open] %s 信号=%s,买入=%d股,原因=反弹已确认",code,signal_key,volume)
def check_timezone(timezone: str, now: datetime | None = None) -> bool:
"""验证当前时间是否处于配置区间。
``*`` 表示全天允许;多个区间用逗号分隔,例如
``9:30-10:30,13:30-14:30``。同时支持跨午夜区间。
"""
timezone = str(timezone or "").strip()
if timezone == "*":
return True
current = now or datetime.now()
current_minutes = current.hour * 60 + current.minute
for section in timezone.split(","):
bounds = section.strip().split("-")
if len(bounds) != 2:
continue
start = _parse_minutes(bounds[0])
end = _parse_minutes(bounds[1])
if start is None or end is None:
continue
if start <= end and start <= current_minutes <= end:
return True
if start > end and (current_minutes >= start or current_minutes <= end):
return True
return False
@lru_cache(maxsize=256)
def _parse_minutes(value: str) -> int | None:
"""把 ``时:分`` 转换为当天分钟数,无效值返回 None。"""
try:
hour_text, minute_text = value.strip().split(":")
hour, minute = int(hour_text), int(minute_text)
except (TypeError, ValueError):
return None
if not 0 <= hour <= 23 or not 0 <= minute <= 59:
return None
return hour * 60 + minute

View File

@@ -1,101 +1,215 @@
"""先卖后买做 T持仓存在 State未买回数量从实际成交计算"""
"""趋势策略持仓止盈与分级补仓"""
import logging as log
import math
from dataclasses import dataclass
from libs.calc import calc_buy_volume, calculate_min_profit_rate
from libs.grid_take_profit import GridState
from sdk import OP_BUY, OP_SELL, PositionItem, Tick
from libs.state import State
from libs.order import PlaceOrderRequest
from libs.runtime import Runtime
from libs.state import State
from sdk import OP_BUY, OP_SELL, PositionItem
import logging as log
LOSS_TIERS = -30.0
def t_rounds(store: State) -> dict[str, dict]:
"""每只证券保留最近一轮 T跨日未买回的数量继续保留。"""
rounds = {}
for deal in sorted(store.deals.values(), key=lambda d: (
d['trade_date'], d['trade_time'].replace(':', ''), d['id']
)):
order = deal['order_local_id']
code = deal['stock_code']
if order.startswith('zt-t-sell-') and deal['offset_flag'] in (24, 49):
item = rounds.get(code)
if item is None or item['bought'] >= item['sold']:
item = rounds[code] = dict(sold=0, bought=0, amount=0.0, date='')
item['sold'] += deal['volume']
item['amount'] += deal['trade_amount']
item['date'] = deal['trade_date']
elif order.startswith('zt-t-buy-') and deal['offset_flag'] in (23, 48) and code in rounds:
rounds[code]['bought'] += deal['volume']
rounds[code]['date'] = deal['trade_date']
return rounds
@dataclass(slots=True)
class TradeDecision:
"""一次止盈或补仓判断的统一结果。"""
submitted: bool
message: str = ""
reserved_cash: float = 0.0
def manage_positions(
run: Runtime, store: State, ticks, positions: list[PositionItem],
rounds: dict[str, dict], available: float, today: str, force_buy_back: bool = False,
) -> float:
"""先偿还买回欠仓;同一证券当天完成一轮后不再卖出。"""
by_code = {p.stock_code: p for p in positions}
codes = dict.fromkeys(list(rounds) + list(store.state))
for code in sorted(codes, key=lambda c: not (c in rounds and rounds[c]['sold'] > rounds[c]['bought'])):
runtime: Runtime,
ticks: dict[str, Tick],
positions: list[PositionItem],
market_ok: bool,
available: float,
state:State,
) -> None:
# 遍历处理每个持仓
for position in positions:
try:
if code in run.account_cfg.excluded_codes:
continue
if run.orders.busy(code, 'BUY') or run.orders.busy(code, 'SELL'):
continue
available = max(0, available)
code = position.stock_code
tick = ticks.get(code)
price = tick.last_price if tick else 0.0
if not math.isfinite(price) or price <= 0:
if code in runtime.account_cfg.excluded_codes:
log.info(
"[Position - ] 代码=%s,名称=%s,止盈=跳过,补仓=跳过,原因=已配置为排除股票",
code,
position.stock_name,
)
continue
item = rounds.get(code)
row = store.state.get(code, {})
position = by_code.get(code)
recorded = row.get('base_qty', 0) + row.get('added_qty', 0)
if recorded != (position.volume if position else 0):
log.warning('[ZT] %s 持仓快照与成交未对齐,等待下一轮', code)
if (
not code
or position.open_price <= 0
or position.volume <= 0
or tick is None
or tick.last_price <= 0
):
log.warning(
"[Position - ] 代码=%s,名称=%s,止盈=跳过,补仓=跳过,原因=持仓或行情数据无效",
code or "未知",
position.stock_name,
)
continue
if item and item['sold'] > item['bought']:
volume = item['sold'] - item['bought']
# 部分成交后的零股欠仓不能按普通买入申报,不扩大买回数量。
minimum = 200 if code.startswith('688') else 100
if not code.startswith('688'):
volume = volume // 100 * 100
if volume < minimum:
log.warning('[ZT] %s 剩余买回 %d 股不满足申报数量,保留欠仓', code, item['sold'] - item['bought'])
continue
target = item['amount'] / item['sold'] * (1 - run.account_cfg.zt_buy_fall_pct / 100)
if not force_buy_back and price > target:
continue
amount = price * volume * 1.01
if amount > available:
log.warning('[ZT] %s 买回资金不足,需要 %.2f,可用 %.2f', code, amount, available)
continue
if not force_buy_back and not run.add_watch.triggered('ZT 买回', code, price):
continue
available -= amount
request = PlaceOrderRequest(OP_BUY, code, volume, run.orders.new_order_id('zt', 't-buy'), 'zt')
if run.orders.place(run.client, request):
run.add_watch.forget(code)
log.info('[ZT 买回] %s %d%s', code, volume, ',尾盘买回' if force_buy_back else '')
posState = state.get_by_code(position.stock_code)
if not posState:
continue
if force_buy_back or (item and item['date'] >= today) or not position or recorded <= 0:
continue
if price > run.account_cfg.zt_max_price:
continue
cost = (row['base_qty'] * row['base_price'] + row['added_qty'] * row['added_price']) / recorded
if cost <= 0:
continue
pnl = (price / cost - 1) * 100
observation = run.profit_tracker.observe(f'{run.account_cfg.account_id}:{code}:{today}', pnl)
if observation.state != GridState.RETREAT or pnl < run.account_cfg.min_profit_pct:
continue
volume = int(min(position.can_use_volume, recorded * run.account_cfg.zt_sell_ratio)) // 100 * 100
if volume < (200 if code.startswith('688') else 100):
continue
request = PlaceOrderRequest(OP_SELL, code, volume, run.orders.new_order_id('zt', 't-sell'), 'zt')
if run.orders.place(run.client, request):
log.info('[ZT 卖出] %s %d 股,按实际成交买回', code, volume)
volume = position.can_use_volume
cost_price = position.open_price
if posState.get('added_qty',0) >=100:
volume = posState.get('added_qty',0)
cost_price = posState.get('added_price',0)
pnl_rate = round(
(tick.last_price - cost_price) / cost_price * 100,
2,
)
minimum_profit = calculate_min_profit_rate(cost_price, 1)
profit_decision = handle_profit(
runtime=runtime,
stock_code=position.stock_code,
volume=volume,
tick=tick,
pnl_rate=pnl_rate,
minimum_profit=minimum_profit,
)
profit_action = profit_decision.message or "未触发"
loss_add_action = "未启用"
if runtime.account_cfg.enable_loss_add_position and market_ok:
loss_decision = handle_loss(
runtime=runtime,
position=position,
tick=tick,
pnl_rate=pnl_rate,
available=available,
)
available = available - loss_decision.reserved_cash
loss_add_action = loss_decision.message or "未触发"
elif runtime.account_cfg.enable_loss_add_position:
loss_add_action = "大盘信号不允许"
strTag = "-"
if pnl_rate >= minimum_profit:
strTag = ""
elif pnl_rate< LOSS_TIERS[0]:
strTag = ""
if strTag != "-":
log.info(
"[Position %s ] %s %s,盈亏=%.2f%%,止盈=%s,补仓=%s",
strTag,
code,
position.stock_name,
pnl_rate,
profit_action,
loss_add_action,
)
except Exception:
log.exception('[ZT 持仓] %s 处理失败', code)
return available
log.exception(
"[Position] 持仓处理异常,代码=%s,继续处理后续持仓",
position.stock_code,
)
def handle_profit(
runtime: Runtime,
stock_code: str,
volume:int,
tick: Tick,
pnl_rate: float,
minimum_profit: float,
) -> TradeDecision:
"""基于跨轮保存的最高盈利网格判断是否提交止盈。"""
if pnl_rate < minimum_profit:
return TradeDecision(False)
key = _position_key(runtime, stock_code)
observation = runtime.profit_tracker.observe(key, pnl_rate)
if observation.state == GridState.ARMED:
return TradeDecision(
False,
f"首次, PNL:{pnl_rate:.2f}%,网格={observation.current_grid}",
)
if observation.state == GridState.RAISED:
return TradeDecision(
False,
f"突破, PNL:{pnl_rate:.2f}%,网格={observation.current_grid}",
)
if observation.state == GridState.STEADY:
return TradeDecision(False,f"持平, PNL:{pnl_rate:.2f}%,网格={observation.current_grid}",)
if runtime.orders.busy(stock_code, "SELL"):
return TradeDecision(False, "卖出委托处理中")
volume = volume % 100
if volume <= 0:
return TradeDecision(False, "无可用整手持仓")
order_id = runtime.orders.new_order_id("zt","SELL")
request = PlaceOrderRequest(
op=OP_SELL,
code=stock_code,
volume=volume,
order_id=order_id,
strategy_name=runtime.account_cfg.strategy,
)
if not runtime.orders.place(runtime.client, request):
return TradeDecision(False, "止盈委托失败")
return TradeDecision(True, f"[止盈卖出] {volume} 股,订单={order_id}")
def handle_loss(
runtime: Runtime,
stock_code: str,
volume:int,
tick: Tick,
pnl_rate: float,
available: float,
) -> TradeDecision:
"""按亏损档位、反弹确认和本轮剩余预算提交补仓。"""
if pnl_rate > LOSS_TIERS:
return TradeDecision(False)
if not runtime.add_watch.triggered("补仓", stock_code, tick.last_price):
return TradeDecision(False, "等待价格反弹确认")
if runtime.orders.busy(stock_code, "BUY"):
return TradeDecision(False, "买入委托处理中")
volume = calc_buy_volume(tick.last_price, runtime.account_cfg.buy_value)
amount = tick.last_price * volume
if volume <= 0 or amount > available:
return TradeDecision(False, "本轮可用资金不足")
order_id = runtime.orders.new_order_id("zt","added")
request = PlaceOrderRequest(
op=OP_BUY,
code=stock_code,
volume=volume,
order_id=order_id,
strategy_name=runtime.account_cfg.strategy,
kind="add",
)
if not runtime.orders.place(runtime.client, request):
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