sanguo_vnpy_v2/sanguo_live/strategies.py

"""A 股 15min 双均线策略(AShareDoubleMaStrategy)。

为什么不直接用 ``vnpy_ctastrategy.strategies.DoubleMaStrategy``:
- 标准 DoubleMa 在 ``on_init`` 里 ``BarGenerator(self.on_bar)`` —— 无 ``window`` 参数,
  只合成 1min bar;且其 ``parameters = ["fast_window", "slow_window"]`` 不暴露周期配置,
  **无法通过 setting 切到 15min**。
- 本策略改用 ``BarGenerator(self._on_1min_bar, window=15, on_window_bar=self.on_bar)``,
  让 ``on_bar`` 直接收 15min bar,通过 ``window`` 参数(默认 15)可配。
- 定寸 + 禁做空由 :class:`AShareCtaTemplate` 保证。
- A 股不能做空,死叉时只平多,不开空(原 DoubleMa 的 ``pos<0 → cover+buy`` 分支删除)。
"""
from __future__ import annotations

import logging

logger = logging.getLogger(__name__)

try:
    from vnpy.trader.utility import BarGenerator, ArrayManager  # type: ignore
    from vnpy.trader.constant import Interval  # type: ignore
    from vnpy.trader.object import BarData, TickData  # type: ignore
    _DEPS_OK = True
except ImportError:  # 本机 dev 容错
    BarGenerator = None  # type: ignore
    ArrayManager = None  # type: ignore
    Interval = None  # type: ignore
    BarData = None  # type: ignore
    TickData = None  # type: ignore
    _DEPS_OK = False

from sanguo_live.base_template import AShareCtaTemplate


class AShareDoubleMaStrategy(AShareCtaTemplate):
    """A 股 双均线 15min 策略。

    金叉(fast 上穿 slow)且无持仓 → 买开 1 手;
    死叉(fast 下穿 slow)且持多 → 卖平 1 手;
    A 股禁做空 → 不开空单(基类 short 已拦截)。
    """

    author = "sanguo_live"

    fast_window: int = 10
    slow_window: int = 20
    # BarGenerator 分钟窗口。15 表示 15min。A 股支持 2/3/5/6/10/15/20/30(必须能整除 60)。
    window: int = 15

    # 把基类的 size/forbid_short 一起暴露,update_setting 才会回填全部字段。
    parameters = ["fast_window", "slow_window", "window", "size", "forbid_short"]
    variables = ["fast_ma0", "fast_ma1", "slow_ma0", "slow_ma1"]

    fast_ma0: float = 0.0
    fast_ma1: float = 0.0
    slow_ma0: float = 0.0
    slow_ma1: float = 0.0

    def on_init(self) -> None:
        self.write_log("AShareDoubleMa 策略初始化")
        # 关键:用 BarGenerator 把 1min bar 合成 window 分钟 bar 后回调 self.on_bar。
        # _on_1min_bar 故意空实现 —— 我们不处理 1min,只让它喂养 BarGenerator。
        self.bg: BarGenerator = BarGenerator(
            self._on_1min_bar,
            window=self.window,
            on_window_bar=self.on_bar,
        )
        self.am: ArrayManager = ArrayManager()
        self.load_bar(10)

    def on_start(self) -> None:
        self.write_log("AShareDoubleMa 策略启动")
        self.put_event()

    def on_stop(self) -> None:
        self.write_log("AShareDoubleMa 策略停止")
        self.put_event()

    def on_tick(self, tick: TickData) -> None:
        """tick 推入 BarGenerator,由其合成 1min 与 window 分钟 bar。"""
        self.bg.update_tick(tick)

    def _on_1min_bar(self, bar: BarData) -> None:
        """1min bar 回调:故意忽略(由 BarGenerator 内部累积合成 15min)。"""
        return

    def on_bar(self, bar: BarData) -> None:
        """15min(或 setting 配置的 window)bar 回调。"""
        self.cancel_all()

        am: ArrayManager = self.am
        am.update_bar(bar)
        if not am.inited:
            return

        fast_ma = am.sma(self.fast_window, array=True)
        self.fast_ma0 = fast_ma[-1]
        self.fast_ma1 = fast_ma[-2]

        slow_ma = am.sma(self.slow_window, array=True)
        self.slow_ma0 = slow_ma[-1]
        self.slow_ma1 = slow_ma[-2]

        cross_over = self.fast_ma0 > self.slow_ma0 and self.fast_ma1 < self.slow_ma1
        cross_below = self.fast_ma0 < self.slow_ma0 and self.fast_ma1 > self.slow_ma1

        if cross_over:
            if self.pos == 0:
                self.buy(bar.close_price, 1)
            # A 股不可做空:原 DoubleMa 的 pos<0 分支(cover+buy)省略
        elif cross_below:
            if self.pos > 0:
                self.sell(bar.close_price, 1)
            # A 股不可做空:原 DoubleMa 的 pos==0 short 分支省略

        self.put_event()


__all__ = ["AShareDoubleMaStrategy"]