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

* 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

czsc/__init__.py

@@ -13,14 +13,14 @@
 from czsc.analyze import CZSC
 from czsc.traders.advanced import CzscAdvancedTrader, create_advanced_trader
 from czsc.utils.ta import SMA, EMA, MACD, KDJ
-from czsc.objects import Freq, Operate, Direction, Signal, Factor, Event, RawBar, NewBar
+from czsc.objects import Freq, Operate, Direction, Signal, Factor, Event, RawBar, NewBar, Position
 from czsc.utils.cache import home_path, get_dir_size, empty_cache_path
 
 
-__version__ = "0.9.6"
+__version__ = "0.9.7"
 __author__ = "zengbin93"
 __email__ = "[email protected]"
-__date__ = "20221221"
+__date__ = "20230123"
 
 
 if envs.get_welcome():

czsc/analyze.py

@@ -204,7 +204,7 @@ def __update_bi(self):
 
         # 查找笔
         if not self.bi_list:
-            # 第一个笔的查找
+            # 第一笔的查找
             fxs = check_fxs(bars_ubi)
             if not fxs:
                 return
@@ -223,29 +223,27 @@ def __update_bi(self):
             self.bars_ubi = bars_ubi_
             return
 
-        last_bi = self.bi_list[-1]
-
-        # 如果上一笔被破坏，将上一笔的bars与bars_ubi进行合并
-        if (last_bi.direction == Direction.Up and bars_ubi[-1].high > last_bi.high) \
-                or (last_bi.direction == Direction.Down and bars_ubi[-1].low < last_bi.low):
-            bars_ubi_a = last_bi.bars[:-1] + [x for x in bars_ubi if x.dt >= last_bi.bars[-1].dt]
-            self.bi_list.pop(-1)
-        else:
-            bars_ubi_a = bars_ubi
-
-        if self.verbose and len(bars_ubi_a) > 100:
-            logger.info(f"czsc_update_bi: {self.symbol} - {self.freq} - {bars_ubi_a[-1].dt} 未完成笔延伸数量: {len(bars_ubi_a)}")
+        if self.verbose and len(bars_ubi) > 100:
+            logger.info(f"{self.symbol} - {self.freq} - {bars_ubi[-1].dt} 未完成笔延伸数量: {len(bars_ubi)}")
 
         if envs.get_bi_change_th() > 0.5 and len(self.bi_list) >= 5:
-            benchmark = min(last_bi.power_price, np.mean([x.power_price for x in self.bi_list[-5:]]))
+            benchmark = min(self.bi_list[-1].power_price, np.mean([x.power_price for x in self.bi_list[-5:]]))
         else:
             benchmark = None
 
-        bi, bars_ubi_ = check_bi(bars_ubi_a, benchmark)
+        bi, bars_ubi_ = check_bi(bars_ubi, benchmark)
         self.bars_ubi = bars_ubi_
         if isinstance(bi, BI):
             self.bi_list.append(bi)
 
+        # 后处理：如果当前笔被破坏，将当前笔的bars与bars_ubi进行合并，并丢弃
+        last_bi = self.bi_list[-1]
+        bars_ubi = self.bars_ubi
+        if (last_bi.direction == Direction.Up and bars_ubi[-1].high > last_bi.high) \
+                or (last_bi.direction == Direction.Down and bars_ubi[-1].low < last_bi.low):
+            self.bars_ubi = last_bi.bars[:-1] + [x for x in bars_ubi if x.dt >= last_bi.bars[-1].dt]
+            self.bi_list.pop(-1)
+
     def update(self, bar: RawBar):
         """更新分析结果
 
@@ -259,7 +257,7 @@ def update(self, bar: RawBar):
             # 当前 bar 是上一根 bar 的时间延伸
             self.bars_raw[-1] = bar
             if len(self.bars_ubi) >= 3:
-                edt = self.bars_ubi[1].dt
+                edt = self.bars_ubi[-2].dt
                 self.bars_ubi = [x for x in self.bars_ubi if x.dt <= edt]
                 last_bars = [x for x in self.bars_raw[-50:] if x.dt > edt]
             else:
@@ -285,6 +283,8 @@ def update(self, bar: RawBar):
 
         # 更新笔
         self.__update_bi()
+
+        # 根据最大笔数量限制完成 bi_list, bars_raw 序列的数量控制
         self.bi_list = self.bi_list[-self.max_bi_num:]
         if self.bi_list:
             sdt = self.bi_list[0].fx_a.elements[0].dt
@@ -295,6 +295,7 @@ def update(self, bar: RawBar):
                     break
             self.bars_raw = self.bars_raw[s_index:]
 
+        # 如果有信号计算函数，则进行信号计算
         if self.get_signals:
             self.signals = self.get_signals(c=self)
         else:

czsc/connectors/qmt_connector.py

@@ -7,8 +7,12 @@
 """
 import pandas as pd
 from typing import List
+from loguru import logger
+from deprecated import deprecated
 from czsc.objects import Freq, RawBar
+from czsc.fsa.im import IM
 from xtquant import xtdata
+from xtquant.xttrader import XtQuantTraderCallback
 
 
 def format_stock_kline(kline: pd.DataFrame, freq: Freq) -> List[RawBar]:
@@ -46,6 +50,54 @@ 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):
+    """获取 QMT K线数据，实盘、回测通用
+
+    :param symbol: 股票代码 例如：300001.SZ
+    :param period: 周期 分笔"tick" 分钟线"1m"/"5m" 日线"1d"
+    :param start_time: 开始时间
+    :param end_time: 结束时间
+    :param count: 数量 -1全部，n: 从结束时间向前数n个
+    :param dividend_type: 除权类型"none" "front" "back" "front_ratio" "back_ratio"
+
+    :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
+    """
+    start_time = pd.to_datetime(start_time).strftime('%Y%m%d%H%M%S')
+    end_time = pd.to_datetime(end_time).strftime('%Y%m%d%H%M%S')
+    if kwargs.get("download_hist", True):
+        xtdata.download_history_data(symbol, period=period, start_time=start_time, end_time=end_time)
+
+    field_list = ['time', 'open', 'high', 'low', 'close', 'volume', 'amount']
+    data = xtdata.get_market_data(field_list, stock_list=[symbol], period=period, count=count,
+                                  dividend_type=dividend_type, start_time=start_time,
+                                  end_time=end_time, fill_data=kwargs.get("fill_data", False))
+
+    df = pd.DataFrame({key: value.values[0] for key, value in 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
+    df = df.dropna()
+
+    if kwargs.get("df", True):
+        return df
+    else:
+        freq_map = {"1m": Freq.F1, "5m": Freq.F5, "1d": Freq.D}
+        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线数据
 
@@ -95,7 +147,7 @@ def get_symbols(step):
     stocks_map = {
         "index": ['000905.SH', '000016.SH', '000300.SH', '000001.SH', '000852.SH',
                   '399001.SZ', '399006.SZ', '399376.SZ', '399377.SZ', '399317.SZ', '399303.SZ'],
-        "stock": stocks.ts_code.to_list(),
+        "stock": stocks,
         "check": ['000001.SZ'],
         "train": stocks[:200],
         "valid": stocks[200:600],
@@ -107,18 +159,114 @@ def get_symbols(step):
     return stocks_map[step]
 
 
-def test_local_kline():
+class TraderCallback(XtQuantTraderCallback):
+    """基础回调类，主要是一些日志和IM通知功能"""
+    def __init__(self, **kwargs):
+        super(TraderCallback, self).__init__()
+        self.kwargs = kwargs
+
+        if kwargs.get('feishu_app_id', None) and kwargs.get('feishu_app_secret', None):
+            self.im = IM(app_id=kwargs['feishu_app_id'], app_secret=kwargs['feishu_app_secret'])
+            self.members = kwargs['feishu_members']
+        else:
+            self.im = None
+            self.members = None
+
+        file_log = kwargs.get('file_log', None)
+        if file_log:
+            logger.add(file_log, rotation='1 day', encoding='utf-8', enqueue=True)
+        self.file_log = file_log
+        logger.info(f"TraderCallback init: {kwargs}")
+
+    def on_disconnected(self):
+        """连接断开"""
+        logger.info("connection lost")
+
+    def on_stock_order(self, order):
+        """委托回报推送
+
+        :param order: XtOrder对象
+        """
+        logger.info(f"on order callback: {order.stock_code} {order.order_status} {order.order_sysid}")
+
+    def on_stock_asset(self, asset):
+        """资金变动推送
+
+        :param asset: XtAsset对象
+        """
+        logger.info(f"on asset callback: {asset.account_id} {asset.cash} {asset.total_asset}")
+
+    def on_stock_trade(self, trade):
+        """成交变动推送
+
+        :param trade: XtTrade对象
+        """
+        logger.info(f"on trade callback: {trade.account_id} {trade.stock_code} {trade.order_id}")
+
+    def on_stock_position(self, position):
+        """持仓变动推送
+
+        :param position: XtPosition对象
+        """
+        logger.info(f"on position callback: {position.stock_code} {position.volume}")
+
+    def on_order_error(self, order_error):
+        """委托失败推送
+
+        :param order_error:XtOrderError 对象
+        """
+        logger.info(f"on order_error callback: {order_error.order_id} {order_error.error_id} {order_error.error_msg}")
+
+    def on_cancel_error(self, cancel_error):
+        """撤单失败推送
+
+        :param cancel_error: XtCancelError 对象
+        """
+        logger.info(f"{cancel_error.order_id} {cancel_error.error_id} {cancel_error.error_msg}")
+
+    def on_order_stock_async_response(self, response):
+        """异步下单回报推送
+
+        :param response: XtOrderResponse 对象
+        """
+        logger.info(f"on_order_stock_async_response: {response.order_id} {response.seq}")
+
+    def on_account_status(self, status):
+        """账户状态变化推送
+
+        :param status: XtAccountStatus 对象
+        """
+        logger.info(f"on_account_status: {status.account_id} {status.account_type} {status.status}")
+
+
+def test_get_kline():
     # 获取所有板块
     slt = xtdata.get_sector_list()
     stocks = xtdata.get_stock_list_in_sector('沪深A股')
 
-    df = get_local_kline(symbol='000001.SZ', period='1m', count=1000, dividend_type='front',
-                         data_dir=r"D:\迅投极速策略交易系统交易终端 华鑫证券QMT实盘\datadir",
-                         start_time='20200427', end_time='20221231', update=True)
+    df = get_kline(symbol='000001.SZ', period='1m', count=1000, dividend_type='front',
+                   start_time='20200427', end_time='20221231')
     assert not df.empty
-    # df = get_local_kline(symbol='000001.SZ', period='5m', count=1000, dividend_type='front',
-    #                      data_dir=r"D:\迅投极速策略交易系统交易终端 华鑫证券QMT实盘\datadir",
-    #                      start_time='20200427', end_time='20221231', update=False)
-    # df = get_local_kline(symbol='000001.SZ', period='1d', count=1000, dividend_type='front',
-    #                      data_dir=r"D:\迅投极速策略交易系统交易终端 华鑫证券QMT实盘\datadir",
-    #                      start_time='20200427', end_time='20221231', update=False)
+    df = get_kline(symbol='000001.SZ', period='5m', count=1000, dividend_type='front',
+                   start_time='20200427', end_time='20221231')
+    assert not df.empty
+    df = get_kline(symbol='000001.SZ', period='1d', count=1000, dividend_type='front',
+                   start_time='20200427', end_time='20221231')
+    assert not df.empty
+
+
+def test_get_symbols():
+    symbols = get_symbols('index')
+    assert len(symbols) > 0
+    symbols = get_symbols('stock')
+    assert len(symbols) > 0
+    symbols = get_symbols('check')
+    assert len(symbols) > 0
+    symbols = get_symbols('train')
+    assert len(symbols) > 0
+    symbols = get_symbols('valid')
+    assert len(symbols) > 0
+    symbols = get_symbols('etfs')
+    assert len(symbols) > 0
+
+

czsc/data/ts.py

@@ -14,7 +14,7 @@
 from typing import List
 from functools import partial
 from loguru import logger
-
+from tenacity import retry, stop_after_attempt, wait_random
 from czsc.objects import RawBar, Freq
 
 
@@ -42,6 +42,7 @@ def __init__(self, token, timeout=30):
         self.__token = token
         self.__timeout = timeout
 
+    @retry(stop=stop_after_attempt(10), wait=wait_random(1, 5))
     def query(self, api_name, fields='', **kwargs):
         if api_name in ['__getstate__', '__setstate__']:
             return pd.DataFrame()

czsc/fsa/__init__.py

@@ -8,6 +8,7 @@
 import requests
 from czsc.fsa.base import request, FeishuApiBase
 from czsc.fsa.spreed_sheets import SpreadSheets
+from czsc.fsa.im import IM
 
 
 def push_text(text: str, key: str) -> None:

czsc/fsa/base.py

@@ -16,11 +16,11 @@
 import time
 import requests
 from loguru import logger
-from tenacity import retry, stop_after_attempt, stop_after_delay, wait_random
+from tenacity import retry, stop_after_attempt, wait_random
 from requests_toolbelt import MultipartEncoder
 
 
-@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)), wait=wait_random(min=1, max=3))
+@retry(stop=stop_after_attempt(10), wait=wait_random(min=3, max=10))
 def request(method, url, headers, payload=None) -> dict:
     """飞书API标准请求