216 lines
7.2 KiB
Python
216 lines
7.2 KiB
Python
"""趋势策略启动器。
|
||
|
||
该模块负责组合 SDK、配置、状态存储和趋势策略组件，供 main.py 调用。
|
||
"""
|
||
|
||
import time
|
||
import logging as log
|
||
from concurrent.futures import Future, ThreadPoolExecutor
|
||
from copy import deepcopy
|
||
from datetime import datetime
|
||
from threading import Lock
|
||
|
||
import config
|
||
from libs.calc import trading_time
|
||
from libs.market import market_allow_open
|
||
from libs.overview import Overview
|
||
from libs.signal import init_signals, SignalItem
|
||
from sdk import Assets, Client, PositionItem
|
||
from libs.grid_take_profit import GridTrailingTracker
|
||
from libs.order import OrderBook
|
||
from libs.watch import DipWatch
|
||
from libs.runtime import Runtime
|
||
from .open import open_signal
|
||
from .positions import manage_positions
|
||
|
||
_collector_lock = Lock()
|
||
_collector_snapshot: tuple[str, Assets, list[PositionItem]] | None = None
|
||
|
||
|
||
def _cache_portfolio(account_id: str, assets: Assets, positions: list[PositionItem]) -> None:
|
||
"""整体替换最新快照，策略线程不执行序列化和网络上报。"""
|
||
global _collector_snapshot
|
||
with _collector_lock:
|
||
_collector_snapshot = (account_id, assets, positions)
|
||
|
||
|
||
def get_collector_snapshot() -> tuple[str, Assets, list[PositionItem]] | None:
|
||
"""供 scheduler 读取；复制在锁外执行，不阻塞下一轮缓存更新。"""
|
||
with _collector_lock:
|
||
snapshot = _collector_snapshot
|
||
return deepcopy(snapshot)
|
||
|
||
|
||
def StartTrend() -> None:
|
||
"""初始化趋势策略，并以 30 秒间隔持续执行。"""
|
||
client = Client(
|
||
config.global_config.qmt_base_url,
|
||
config.global_config.qmt_token,
|
||
config.HTTP_TIMEOUT,
|
||
)
|
||
executor = None
|
||
try:
|
||
portfolio = client.portfolio()
|
||
assets = portfolio.assets
|
||
positions = list(portfolio.positions.values())
|
||
_cache_portfolio(config.account_config.account_id, assets, positions)
|
||
order_book = OrderBook("trend")
|
||
order_book.refresh(client, portfolio.orders)
|
||
|
||
# 获取本策略的信号开仓数据
|
||
signals = init_signals(
|
||
config.global_config,
|
||
config.account_config.signal_allow,
|
||
)
|
||
log.info(
|
||
"[启动] Trend策略已启动，账户=%s，信号=%d，持仓=%d",
|
||
config.account_config.account_id,
|
||
len(signals),
|
||
len(positions),
|
||
)
|
||
executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="trend")
|
||
run = Runtime(
|
||
client=client,
|
||
global_cfg=config.global_config,
|
||
account_cfg=config.account_config,
|
||
orders=order_book,
|
||
open_watch=DipWatch(),
|
||
add_watch=DipWatch(),
|
||
profit_tracker=GridTrailingTracker(config.account_config.grid_step_pct),
|
||
executor=executor,
|
||
)
|
||
|
||
Overview(assets, positions, config.account_config)
|
||
|
||
DEFAULT_TICK_INTERVAL = 30
|
||
while True:
|
||
lt = time.localtime()
|
||
if (lt.tm_hour, lt.tm_min, lt.tm_sec) >= (15, 0, 0):
|
||
log.info("[Trend] 已到 15:00，结束趋势策略")
|
||
return
|
||
current_sec = lt.tm_sec
|
||
|
||
# 计算距离下一个目标时间点（0秒或30秒）的等待时间
|
||
if current_sec < DEFAULT_TICK_INTERVAL:
|
||
wait_seconds = DEFAULT_TICK_INTERVAL - current_sec
|
||
elif current_sec < 60:
|
||
wait_seconds = 60 - current_sec
|
||
else:
|
||
wait_seconds = DEFAULT_TICK_INTERVAL
|
||
|
||
# 等待到目标时间点
|
||
time.sleep(wait_seconds)
|
||
|
||
# 单轮失败不能杀死唯一的交易定时线程。
|
||
try:
|
||
RunOnce(run, signals)
|
||
except Exception as e:
|
||
log.error(
|
||
f"[Trend] 本 tick 执行失败，下一 tick 继续: {e}", exc_info=True
|
||
)
|
||
finally:
|
||
try:
|
||
if executor is not None:
|
||
executor.shutdown(wait=True)
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def RunOnce(run: Runtime, signals: list[SignalItem]) -> None:
|
||
"""按固定步骤执行一轮趋势策略， ``RunOnce``。"""
|
||
if not trading_time(datetime.now()):
|
||
return
|
||
|
||
print(
|
||
"=" * 40 + f" Ticker {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} " + "=" * 40
|
||
)
|
||
|
||
started_at = time.monotonic()
|
||
|
||
# 1. 一次获取资产、持仓和订单，并清理过期订单。
|
||
try:
|
||
portfolio = run.client.portfolio()
|
||
assets = portfolio.assets
|
||
position_codes = list(portfolio.positions)
|
||
positions = list(portfolio.positions.values())
|
||
_cache_portfolio(run.account_cfg.account_id, assets, positions)
|
||
run.orders.refresh(run.client, portfolio.orders)
|
||
except Exception:
|
||
log.exception("[Portfolio] 刷新账户快照失败")
|
||
return
|
||
|
||
futures: list[tuple[str, Future]] = []
|
||
|
||
# 2. 验证可用资金；低于资金安全线时禁止开新仓。
|
||
allow_open_by_cash = (
|
||
assets.available >= assets.total * run.account_cfg.min_cash_ratio
|
||
)
|
||
if not allow_open_by_cash:
|
||
log.info(
|
||
"[Status] 禁止开仓：可用资金不足，可用=%.2f，总资产=%.2f",
|
||
assets.available,
|
||
assets.total,
|
||
)
|
||
|
||
# 3. 获取大盘状态，只有大盘信号允许时才执行开仓。
|
||
market_ok = market_allow_open()
|
||
|
||
# 4. 验证有效开仓信号：排除已有持仓和未决订单。
|
||
allow_open: list[SignalItem] = []
|
||
allow_codes: list[str] = []
|
||
for signal in signals:
|
||
if signal.code not in portfolio.positions:
|
||
allow_open.append(signal)
|
||
allow_codes.append(signal.code)
|
||
|
||
if allow_open and not market_ok:
|
||
log.info("[开仓] 禁止开仓：大盘信号不允许，候选=%d", len(allow_open))
|
||
|
||
# 5. 获取持仓和待开仓证券的实时行情 tick。
|
||
all_codes = list(dict.fromkeys(position_codes + allow_codes))
|
||
try:
|
||
ticks = run.client.full_tick(all_codes)
|
||
except Exception:
|
||
log.exception("[行情] 获取行情失败，代码数量=%d", len(all_codes))
|
||
return
|
||
|
||
log.info(
|
||
"[RunOnce] 本轮就绪，持仓=%d，候选=%d，大盘允许=%s，资金允许=%s",
|
||
len(positions),
|
||
len(allow_open),
|
||
market_ok,
|
||
allow_open_by_cash,
|
||
)
|
||
|
||
# 启动线程，开始计算
|
||
# 7. 持仓计算。
|
||
futures.append(
|
||
(
|
||
"持仓计算",
|
||
run.executor.submit(
|
||
manage_positions, run, ticks, positions, market_ok, assets.available
|
||
),
|
||
)
|
||
)
|
||
|
||
# 8. 开仓计算：必须同时存在有效信号且大盘允许开仓。
|
||
if allow_open and market_ok and allow_open_by_cash:
|
||
futures.append(
|
||
("开仓计算", run.executor.submit(open_signal, run, ticks, allow_open))
|
||
)
|
||
|
||
# 9. 开始执行
|
||
for name, future in futures:
|
||
_wait_worker(name, future)
|
||
log.info(
|
||
"[RunOnce] 本轮完成，耗时=%d毫秒", int((time.monotonic() - started_at) * 1000)
|
||
)
|
||
|
||
|
||
def _wait_worker(name: str, future: Future) -> None:
|
||
"""保留单轮继续运行的语义，分别记录工作线程异常。"""
|
||
try:
|
||
future.result()
|
||
except Exception:
|
||
log.exception("[运行] %s线程失败", name)
|