big-qmt/py-client/strategy/trend/open.py

"""趋势策略开仓逻辑。"""

from __future__ import annotations

from datetime import datetime

from libs import calc_buy_volume
from sdk import OP_BUY
from .runtime import Runtime
from .order import PlaceOrderRequest
import logging as log


def open_signal(run:Runtime, ticks, open_signals) -> None:
    """逐个验证开仓信号并提交买入委托。"""
    for item in open_signals:
        if item.code in run.account_cfg.excluded_codes:
            log.info("[Open] %s 信号=%s，跳过：已配置为排除股票", item.code, item.signal_key)
            continue
        # 1. 验证信号配置允许开仓的时间区间。
        signal_config = run.global_cfg.signals.get(item.signal_key)
        if signal_config is None:
            log.warning(
                "[Open] %s 信号=%s，跳过：未找到信号配置",
                item.code,
                item.signal_key,
            )
            continue
        if not check_timezone(signal_config.timezone):
            log.info(
                "[Open] %s 信号=%s，跳过：不在信号时间段(%s)",
                item.code,
                item.signal_key,
                signal_config.timezone,
            )
            continue

        # 2. 检查该证券是否已有买入委托锁，防止重复下单。
        if run.orders.busy(item.code,"BUY"):
            log.info("[Open] %s 信号=%s，跳过：买入委托处理中", item.code, item.signal_key)
            continue

        # 3. 验证行情和最新价格是否有效。
        tick = ticks.get(item.code)
        price = tick.last_price if tick is not None else 0
        if price <= 0:
            log.info("[Open] %s 信号=%s，跳过：价格无效", item.code, item.signal_key)
            continue

        # 5. 根据单笔买入金额计算整手开仓数量。
        volume = calc_buy_volume(price, run.account_cfg.buy_value)
        if volume <= 0:
            log.info("[Open] %s 信号=%s，跳过：数量无效", item.code, item.signal_key)
            continue

        # 当前价高于昨收价可开仓
        if signal_config.gt_last_price_is_open and item.last_close>0 and price>item.last_close:
            try:
                do_open(run, item.code, volume, item.signal_key, price)
                log.info("[Open] %s 信号=%s，买入=%d股，原因=现价高于昨收", item.code, item.signal_key, volume)
            except RuntimeError as exc:
                log.info("[Open] %s 信号=%s，买入=%d股失败：%s", item.code, item.signal_key, volume, exc)
            except Exception:
                log.exception("[Open] %s 信号=%s，买入=%d股异常", item.code, item.signal_key, volume)
            continue

        # 4. 等待价格从观察低点反弹，防止直接接下跌中的“飞刀”。
        if not run.open_watch.triggered("开仓", item.code, price):
            continue

        try:
            do_open(run, item.code, volume, item.signal_key, price)
            log.info("[Open] %s 信号=%s，买入=%d股，原因=反弹已确认", item.code, item.signal_key, volume)
        except RuntimeError as exc:
            log.warning("[Open] %s 信号=%s，买入=%d股失败：%s", item.code, item.signal_key, volume, exc)
        except Exception:
            log.exception("[Open] %s 信号=%s，买入=%d股异常", item.code, item.signal_key, volume)
        

def do_open(run: Runtime, code: str, volume: int, signal_key: str, price: float) -> None:
    """生成本地订单号并按最新价提交开仓委托。"""
    order_id = run.orders.new_order_id()
    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)


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


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