V0.9.7 更新代码 (waditu#122)

* 0.9.6 update * 0.9.6 update * 0.9.7 新增 tas_kdj_evc_V221201 信号 * 0.9.7 新增 tas_kdj_evc_V221201 信号 * 0.9.7 新增飞书消息发送 * 0.9.7 新增结合大盘择时或股票池的择时优化分析 * 0.9.7 update * 0.9.7 fix update_bi * 0.9.7 update * 0.9.7 update * 0.9.7 CzscTrader 新增 on_sig 模式 * 0.9.7 fix * 0.9.7 update * 0.9.7 增加持仓评价 * 0.9.7 支持自定义仓位集成方法 * 0.9.7 支持自定义仓位集成方法 * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 fix test * 0.9.7 fix test * 0.9.7 update * 0.9.7 deprecated * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 update * 0.9.7 fix bug * 0.9.7 update * 0.9.7 update * 0.9.7 新增 QmtTradeManager 用于实盘仿真 * 0.9.7 新增 QmtTradeManager 用于实盘仿真 * 0.9.7 update examples * 0.9.7 fix test
cpcjcey · Feb 10, 2023 · 596342d · 596342d
1 parent 9dba6a7
commit 596342d
Show file tree

Hide file tree

Showing 30 changed files with 673 additions and 351 deletions.
diff --git a/czsc/analyze.py b/czsc/analyze.py
@@ -373,6 +373,6 @@ def fx_list(self) -> List[FX]:
             fxs.extend(bi_.fxs[1:])
         ubi = self.ubi_fxs
         for x in ubi:
-            if x.dt > fxs[-1].dt:
+            if not fxs or x.dt > fxs[-1].dt:
                 fxs.append(x)
         return fxs
diff --git a/czsc/connectors/qmt_connector.py b/czsc/connectors/qmt_connector.py
@@ -3,16 +3,23 @@
 author: zengbin93
 email: [email protected]
 create_dt: 2022/12/31 16:03
-describe: 
+describe: QMT 量化交易平台接口
 """
+import time
+import random
 import pandas as pd
 from typing import List
+from tqdm import tqdm
 from loguru import logger
 from deprecated import deprecated
+from datetime import datetime, timedelta
 from czsc.objects import Freq, RawBar
 from czsc.fsa.im import IM
+from czsc.traders.base import CzscTrader
+from xtquant import xtconstant
 from xtquant import xtdata
-from xtquant.xttrader import XtQuantTraderCallback
+from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback
+from xtquant.xttype import StockAccount
 
 
 def format_stock_kline(kline: pd.DataFrame, freq: Freq) -> List[RawBar]:
@@ -50,7 +57,7 @@ def format_stock_kline(kline: pd.DataFrame, freq: Freq) -> List[RawBar]:
     return bars
 
 
-def get_kline(symbol, period, start_time, end_time, count=-1, dividend_type='none', **kwargs):
+def get_kline(symbol, period, start_time, end_time, count=-1, dividend_type='front_ratio', **kwargs):
     """获取 QMT K线数据，实盘、回测通用
 
     :param symbol: 股票代码 例如：300001.SZ
@@ -97,46 +104,6 @@ def get_kline(symbol, period, start_time, end_time, count=-1, dividend_type='non
         return format_stock_kline(df, freq=freq_map[period])
 
 
-@deprecated(version='1.0.0', reason="已废弃，请使用 get_kline 代替")
-def get_local_kline(symbol, period, start_time, end_time, count=-1, dividend_type='none', data_dir=None, update=True):
-    """获取 QMT 本地K线数据
-
-    :param symbol: 股票代码 例如：'300001.SZ'
-    :param period: 周期 分笔"tick" 分钟线"1m"/"5m" 日线"1d"
-    :param start_time: 开始时间，格式YYYYMMDD/YYYYMMDDhhmmss/YYYYMMDDhhmmss.milli，
-        例如："20200427" "20200427093000" "20200427093000.000"
-    :param end_time: 结束时间 格式同上
-    :param count: 数量 -1全部，n: 从结束时间向前数n个
-    :param dividend_type: 除权类型"none" "front" "back" "front_ratio" "back_ratio"
-    :param data_dir: 下载QMT本地数据路径，如 D:/迅投极速策略交易系统交易终端/datadir
-    :param update: 更新QMT本地数据路径中的数据
-    :return: df Dataframe格式的数据，样例如下
-                         time   open   high    low  close  volume      amount  \
-        0 2022-12-01 10:15:00  13.22  13.22  13.16  13.18   20053  26432861.0
-        1 2022-12-01 10:20:00  13.18  13.19  13.15  13.15   32667  43002512.0
-        2 2022-12-01 10:25:00  13.16  13.18  13.13  13.16   32466  42708049.0
-        3 2022-12-01 10:30:00  13.16  13.19  13.13  13.18   15606  20540461.0
-        4 2022-12-01 10:35:00  13.20  13.25  13.19  13.22   29959  39626170.0
-              symbol
-        0  000001.SZ
-        1  000001.SZ
-        2  000001.SZ
-        3  000001.SZ
-        4  000001.SZ
-    """
-    field_list = ['time', 'open', 'high', 'low', 'close', 'volume', 'amount']
-    if update:
-        xtdata.download_history_data(symbol, period, start_time='20100101', end_time='21000101')
-    local_data = xtdata.get_local_data(field_list, [symbol], period, count=count, dividend_type=dividend_type,
-                                       start_time=start_time, end_time=end_time, data_dir=data_dir)
-
-    df = pd.DataFrame({key: value.values[0] for key, value in local_data.items()})
-    df['time'] = pd.to_datetime(df['time'], unit='ms') + pd.to_timedelta('8H')
-    df.reset_index(inplace=True, drop=True)
-    df['symbol'] = symbol
-    return df
-
-
 def get_symbols(step):
     """获取择时策略投研不同阶段对应的标的列表
 
@@ -239,6 +206,209 @@ def on_account_status(self, status):
         logger.info(f"on_account_status: {status.account_id} {status.account_type} {status.status}")
 
 
+class QmtTradeManager:
+    """QMT交易管理器"""
+
+    def __init__(self, mini_qmt_dir, account_id, **kwargs):
+        """
+
+        :param mini_qmt_dir: mini QMT 路径；如 D:\\国金QMT交易端模拟\\userdata_mini
+        :param account_id: 账户ID
+        :param kwargs:
+
+        """
+        self.symbols = kwargs.get('symbols', [])  # 交易标的列表
+        self.strategy = kwargs.get('strategy', [])  # 交易策略
+        self.symbol_max_pos = kwargs.get('symbol_max_pos', 0.5)  # 每个标的最大持仓比例
+        self.mini_qmt_dir = mini_qmt_dir
+        self.account_id = account_id
+        self.period = kwargs.get('period', '5m')
+        self.delta_days = int(kwargs.get('delta_days', 1))  # 定时执行获取的K线天数
+
+        self.session = random.randint(10000, 20000)
+        self.xtt = XtQuantTrader(mini_qmt_dir, session=self.session, callback=TraderCallback())
+        self.acc = StockAccount(account_id, 'STOCK')
+        self.xtt.start()
+        self.xtt.connect()
+        assert self.xtt.connected, "交易服务器连接失败"
+        _res = self.xtt.subscribe(self.acc)
+        assert _res == 0, "账号订阅失败"
+        self.traders = self.__create_traders(**kwargs)
+
+    def __create_traders(self, **kwargs):
+        """创建交易策略"""
+        traders = {}
+        for symbol in tqdm(self.symbols, desc="创建交易对象", unit="个"):
+            try:
+                bars = get_kline(symbol, self.period, '20180201', datetime.now(),
+                                 dividend_type='front_ratio', df=False, download_hist=True)
+                trader: CzscTrader = self.strategy(symbol=symbol, data_source='qmt').init_trader(bars, sdt='20220601')
+                traders[symbol] = trader
+            except Exception as e:
+                logger.warning(f'创建交易对象失败，symbol={symbol}, e={e}')
+        return traders
+
+    def get_assets(self):
+        """获取账户资产"""
+        return self.xtt.query_stock_asset(self.acc)
+
+    def query_stock_orders(self, cancelable_only=False):
+        """查询股票市场的委托单
+
+        http://docs.thinktrader.net/pages/ee0e9b/#%E5%A7%94%E6%89%98%E6%9F%A5%E8%AF%A2
+
+        :param cancelable_only:
+        :return:
+        """
+        return self.xtt.query_stock_orders(self.acc, cancelable_only)
+
+    def is_order_exist(self, symbol, order_type, volume=None):
+        """判断是否存在相同的委托单"""
+        orders = self.query_stock_orders(cancelable_only=True)
+        for o in orders:
+            if o.stock_code == symbol and o.order_type == order_type:
+                if not volume or o.order_volume == volume:
+                    return True
+        return False
+
+    def is_allow_open(self, symbol, price):
+        """判断是否允许开仓
+
+        :param symbol: 股票代码
+        :param price: 股票现价
+        :return: True 允许开仓，False 不允许开仓
+        """
+        # 如果 未成交的开仓委托单 存在，不允许开仓
+        if self.is_order_exist(symbol, order_type=23):
+            logger.warning(f"存在未成交的开仓委托单，symbol={symbol}")
+            return False
+
+        # 如果 symbol_max_pos 为 0，不允许开仓
+        if self.symbol_max_pos <= 0:
+            return False
+
+        # 如果已经有持仓，不允许开仓
+        if self.query_stock_positions().get(symbol, None):
+            return False
+
+        # 如果资金不足，不允许开仓
+        assets = self.get_assets()
+        if assets.cash < price * 120:
+            logger.warning(f"资金不足，无法开仓，symbol={symbol}")
+            return False
+
+        return True
+
+    def query_stock_positions(self):
+        """查询股票市场的持仓单
+
+        http://docs.thinktrader.net/pages/ee0e9b/#%E6%8C%81%E4%BB%93%E6%9F%A5%E8%AF%A2
+        """
+        res = self.xtt.query_stock_positions(self.acc)
+        if len(res) > 0:
+            res = {x.stock_code: x for x in res}
+        else:
+            res = {}
+        return res
+
+    def send_stock_order(self, **kwargs):
+        """股票市场交易下单
+
+        http://docs.thinktrader.net/pages/ee0e9b/#%E8%82%A1%E7%A5%A8%E5%90%8C%E6%AD%A5%E6%8A%A5%E5%8D%95
+        http://docs.thinktrader.net/pages/198696/#%E6%8A%A5%E4%BB%B7%E7%B1%BB%E5%9E%8B-price-type
+
+        stock_code: 证券代码, 例如"600000.SH"
+        order_type: 委托类型, 23:买, 24:卖
+        order_volume: 委托数量, 股票以'股'为单位, 债券以'张'为单位
+        price_type: 报价类型, 详见帮助手册
+            xtconstant.LATEST_PRICE	5	最新价
+            xtconstant.FIX_PRICE	11	限价
+        price: 报价价格, 如果price_type为限价, 那price为指定的价格, 否则填0
+        strategy_name: 策略名称
+        order_remark: 委托备注
+
+        :return: 返回下单请求序号, 成功委托后的下单请求序号为大于0的正整数, 如果为-1表示委托失败
+        """
+        stock_code = kwargs.get('stock_code')
+        order_type = kwargs.get('order_type')
+        order_volume = kwargs.get('order_volume')
+        price_type = kwargs.get('price_type', xtconstant.LATEST_PRICE)
+        price = kwargs.get('price', 0)
+        strategy_name = kwargs.get('strategy_name', "程序下单")
+        order_remark = kwargs.get('order_remark', "程序下单")
+
+        if not self.xtt.connected:
+            self.xtt.connect()
+            self.xtt.start()
+
+        if order_volume % 100 != 0:
+            order_volume = order_volume // 100 * 100
+
+        assert self.xtt.connected, "交易服务器连接断开"
+        _id = self.xtt.order_stock(self.acc, stock_code, order_type, order_volume,
+                                   price_type, price, strategy_name, order_remark)
+        return _id
+
+    def update_traders(self):
+        """更新交易策略"""
+        holds = self.query_stock_positions()
+        kline_sdt = datetime.now() - timedelta(days=self.delta_days)
+
+        for symbol in self.traders.keys():
+            try:
+                trader = self.traders[symbol]
+                bars = get_kline(symbol, self.period, kline_sdt, datetime.now(),
+                                 dividend_type='front_ratio', df=False, download_hist=True)
+
+                news = [x for x in bars if x.dt > trader.end_dt]
+                if news:
+                    logger.info(f"{symbol} 需要更新的K线数量：{len(news)} | 最新的K线时间是 {news[-1].dt}")
+                    for bar in news:
+                        trader.on_bar(bar)
+
+                    # 根据策略的交易信号，下单【股票只有多头】
+                    if trader.get_ensemble_pos(method='vote') == 1 and self.is_allow_open(symbol, price=news[-1].close):
+                        assets = self.get_assets()
+                        order_volume = min(self.symbol_max_pos * assets.total_asset, assets.cash) // news[-1].close
+                        self.send_stock_order(stock_code=symbol, order_type=23, order_volume=order_volume)
+
+                    # 平多头
+                    if trader.get_ensemble_pos(method='vote') == 0 and symbol in holds.keys():
+                        order_volume = holds[symbol].can_use_volume
+                        self.send_stock_order(stock_code=symbol, order_type=24, order_volume=order_volume)
+
+                    # 更新交易对象
+                    self.traders[symbol] = trader
+                else:
+                    logger.info(f"{symbol} 没有需要更新的K线，最新的K线时间是 {trader.end_dt}")
+
+            except Exception as e:
+                logger.error(f"{symbol} 更新交易策略失败，原因是 {e}")
+
+    def run(self, mode='30m'):
+        """运行策略"""
+        if mode.lower() == '15m':
+            _times = ["09:45", "10:00", "10:15", "10:30", "10:45", "11:00", "11:15", "11:30",
+                      "13:15", "13:30", "13:45", "14:00", "14:15", "14:30", "14:45", "15:00"]
+        elif mode.lower() == '30m':
+            _times = ["09:45", "10:00", "10:30", "11:00", "11:30", "13:30", "14:00", "14:30", "15:00"]
+        elif mode.lower() == '60m':
+            _times = ["10:30", "11:30", "13:45", "14:30"]
+        else:
+            raise ValueError("mode 只能是 15m, 30m, 60m")
+
+        while 1:
+            if datetime.now().strftime("%H:%M") in _times:
+                self.update_traders()
+            else:
+                time.sleep(3)
+
+            # 如果断开，重新连接交易服务器
+            if not self.xtt.connected:
+                self.xtt.connect()
+                self.xtt.start()
+
+
 def test_get_kline():
     # 获取所有板块
     slt = xtdata.get_sector_list()

diff --git a/czsc/eda.py b/czsc/eda.py
@@ -0,0 +1,32 @@
+# -*- coding: utf-8 -*-
+"""
+author: zengbin93
+email: [email protected]
+create_dt: 2023/2/7 13:17
+describe: 用于探索性分析的函数
+"""
+import numpy as np
+import pandas as pd
+
+
+def vwap(price: np.array, volume: np.array, **kwargs) -> float:
+    """计算成交量加权平均价
+    
+    :param price: 价格序列
+    :param volume: 成交量序列
+    :return: 平均价
+    """
+    return np.average(price, weights=volume)
+
+
+def twap(price: np.array, **kwargs) -> float:
+    """计算时间加权平均价
+
+    :param price: 价格序列
+    :return: 平均价
+    """
+    return np.average(price)
+
+
+
+
diff --git a/czsc/objects.py b/czsc/objects.py
@@ -9,6 +9,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from loguru import logger
+from deprecated import deprecated
 from typing import List, Callable
 from transitions import Machine
 from czsc.enum import Mark, Direction, Freq, Operate
@@ -599,6 +600,7 @@ def cal_break_even_point(seq: List[float]) -> float:
     return sub_i / len(seq)
 
 
+@deprecated(reason="择时策略将使用 Position + CzscTrader 代替")
 def evaluate_pairs(pairs, symbol: str, trade_dir: str, cost: float = 0.003) -> dict:
     """评估交易表现
 
@@ -648,6 +650,7 @@ def evaluate_pairs(pairs, symbol: str, trade_dir: str, cost: float = 0.003) -> d
     return p
 
 
+@deprecated(reason="择时策略将使用 Position + CzscTrader 代替")
 class PositionLong:
     def __init__(self, symbol: str,
                  hold_long_a: float = 0.5,
@@ -827,6 +830,7 @@ def update(self, dt: datetime, op: Operate, price: float, bid: int, op_desc: str
             self.long_bid = -1.0
 
 
+@deprecated(reason="择时策略将使用 Position + CzscTrader 代替")
 class PositionShort:
     def __init__(self, symbol: str,
                  hold_short_a: float = 0.5,
@@ -1052,6 +1056,7 @@ def __two_operates_pair(self, op1, op2):
         assert op1['op'] in [Operate.LO, Operate.SO]
         pair = {
             '标的代码': self.symbol,
+            '策略标记': self.name,
             '交易方向': "多头" if op1['op'] == Operate.LO else "空头",
             '开仓时间': op1['dt'],
             '平仓时间': op2['dt'],
@@ -1115,7 +1120,7 @@ def evaluate_pairs(self, trade_dir: str = "多空") -> dict:
             pairs = self.pairs
         else:
             pairs = [x for x in self.pairs if x['交易方向'] == trade_dir]
-        p = {"交易标的": self.symbol, "交易方向": trade_dir,
+        p = {"交易标的": self.symbol, "策略标记": self.name, "交易方向": trade_dir,
              "交易次数": len(pairs), '累计收益': 0, '单笔收益': 0,
              '盈利次数': 0, '累计盈利': 0, '单笔盈利': 0,
              '亏损次数': 0, '累计亏损': 0, '单笔亏损': 0,

diff --git a/czsc/signals/tas.py b/czsc/signals/tas.py
@@ -317,7 +317,7 @@ def tas_macd_first_bs_V221201(c: CZSC, di: int = 1, **kwargs):
             v1 = "一买"
 
         s1_con1 = len(cross) > 3 and cross[-1]['类型'] == '金叉' and cross[-1]['慢线'] > 0
-        s1_con2 = len(dn) > 3 and up[-2]['慢线'] > 0 and up[-3]['慢线'] > 0
+        s1_con2 = len(up) > 3 and up[-2]['慢线'] > 0 and up[-3]['慢线'] > 0
         s1_con3 = len(macd) > 10 and macd[-1] < macd[-2]
         if s1_con1 and s1_con2 and s1_con3:
             v1 = "一卖"