From b0ee0d03e60dfd285a7106706cca2c33a55c4d24 Mon Sep 17 00:00:00 2001 From: zengbin93 Date: Fri, 13 Oct 2023 10:52:48 +0800 Subject: [PATCH 1/5] 0.9.32 start coding --- .github/workflows/pythonpackage.yml | 2 +- czsc/__init__.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml index c359b073b..03943908f 100644 --- a/.github/workflows/pythonpackage.yml +++ b/.github/workflows/pythonpackage.yml @@ -5,7 +5,7 @@ name: Python package on: push: - branches: [ master, V0.9.31 ] + branches: [ master, V0.9.32 ] pull_request: branches: [ master ] diff --git a/czsc/__init__.py b/czsc/__init__.py index 3a70a283c..d11806684 100644 --- a/czsc/__init__.py +++ b/czsc/__init__.py @@ -101,10 +101,10 @@ normalize_feature, ) -__version__ = "0.9.31" +__version__ = "0.9.32" __author__ = "zengbin93" __email__ = "zeng_bin8888@163.com" -__date__ = "20231007" +__date__ = "20231013" From f4c5b60fc8d0288be6684a85ec49dbd8ad4dc9d7 Mon Sep 17 00:00:00 2001 From: zengbin93 Date: Sat, 14 Oct 2023 16:37:59 +0800 Subject: [PATCH 2/5] =?UTF-8?q?0.9.32=20=E4=BC=98=E5=8C=96streamlit?= =?UTF-8?q?=E5=88=86=E6=9E=90=E7=BB=84=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- czsc/utils/st_components.py | 108 +++++++++++++++++++++++++++++------- 1 file changed, 89 insertions(+), 19 deletions(-) diff --git a/czsc/utils/st_components.py b/czsc/utils/st_components.py index a0ca2050f..2d8316bb6 100644 --- a/czsc/utils/st_components.py +++ b/czsc/utils/st_components.py @@ -1,4 +1,5 @@ import czsc +import numpy as np import pandas as pd import streamlit as st import plotly.express as px @@ -8,30 +9,40 @@ def show_daily_return(df, **kwargs): """用 streamlit 展示日收益""" assert df.index.dtype == 'datetime64[ns]', "index必须是datetime64[ns]类型, 请先使用 pd.to_datetime 进行转换" - type_ = "持有日" if kwargs.get("none_zero", False) else "交易日" - df = df.copy().fillna(0) - stats = [] - for col in df.columns: - col_stats = czsc.daily_performance([x for x in df[col] if x != 0]) if type_ == '持有日' else czsc.daily_performance(df[col]) - col_stats['日收益名称'] = col - stats.append(col_stats) - - stats = pd.DataFrame(stats).set_index('日收益名称') - fmt_cols = ['年化', '夏普', '最大回撤', '卡玛', '年化波动率', '非零覆盖', '日胜率', '盈亏平衡点'] - stats = stats.style.background_gradient(cmap='RdYlGn_r', axis=None).format('{:.4f}', subset=fmt_cols) - - df = df.cumsum() - fig = px.line(df, y=df.columns.to_list(), title="日收益累计曲线") - for col in kwargs.get("legend_only_cols", []): - fig.update_traces(visible="legendonly", selector=dict(name=col)) - - title = kwargs.get("title", "") + + def _stats(df_, type_='持有日'): + df_ = df_.copy() + stats = [] + for col in df_.columns: + if type_ == '持有日': + col_stats = czsc.daily_performance([x for x in df_[col] if x != 0]) + else: + assert type_ == '交易日', "type_ 参数必须是 持有日 或 交易日" + col_stats = czsc.daily_performance(df_[col]) + col_stats['日收益名称'] = col + stats.append(col_stats) + stats = pd.DataFrame(stats).set_index('日收益名称') + fmt_cols = ['年化', '夏普', '最大回撤', '卡玛', '年化波动率', '非零覆盖', '日胜率', '盈亏平衡点'] + stats = stats.style.background_gradient(cmap='RdYlGn_r', axis=None).format('{:.4f}', subset=fmt_cols) + return stats + with st.container(): + title = kwargs.get("title", "") if title: st.subheader(title) st.divider() - st.dataframe(stats, use_container_width=True) + + col1, col2 = st.columns([1, 1]) + col1.write("交易日绩效指标") + col1.dataframe(_stats(df, type_='交易日'), use_container_width=True) + col2.write("持有日绩效指标") + col2.dataframe(_stats(df, type_='持有日'), use_container_width=True) + + df = df.cumsum() + fig = px.line(df, y=df.columns.to_list(), title="日收益累计曲线") + for col in kwargs.get("legend_only_cols", []): + fig.update_traces(visible="legendonly", selector=dict(name=col)) st.plotly_chart(fig, use_container_width=True) @@ -143,3 +154,62 @@ def _layering(x): st.write(f"多头：{long}，空头：{short}") mrr['多空组合'] = (mrr[long] - mrr[short]) / 2 show_daily_return(mrr[['多空组合']]) + + +def show_symbol_factor_layering(df, x_col, y_col='n1b', **kwargs): + """使用 streamlit 绘制单个标的上的因子分层收益率图 + + :param df: 因子数据，必须包含 dt, x_col, y_col 列，其中 dt 为日期，x_col 为因子值，y_col 为收益率，数据样例： + + =================== ============ ============ + dt intercept n1b + =================== ============ ============ + 2017-01-03 00:00:00 0 0.00716081 + 2017-01-04 00:00:00 -0.00154541 0.000250816 + 2017-01-05 00:00:00 0.000628884 -0.0062695 + 2017-01-06 00:00:00 -0.00681021 0.00334212 + 2017-01-09 00:00:00 0.00301077 -0.00182963 + =================== ============ ============ + + :param x_col: 因子列名 + :param y_col: 收益列名 + :param kwargs: + + - n: 分层数量，默认为10 + + """ + df = df.copy() + n = kwargs.get("n", 10) + if df[y_col].max() > 100: # 如果收益率单位为BP, 转换为万分之一 + df[y_col] = df[y_col] / 10000 + + if df[x_col].nunique() > n: + df[f'{x_col}分层'] = pd.qcut(df[x_col], q=n, labels=False, duplicates='drop') + else: + # 如果因子值的取值数量小于分层数量，直接使用因子独立值排序作为分层 + x_rank = sorted(df[x_col].unique()) + x_rank = {x_rank[i]: f'第{str(i+1).zfill(2)}层' for i in range(len(x_rank))} + st.success(f"因子值分层对应关系：{x_rank}") + df[f'{x_col}分层'] = df[x_col].apply(lambda x: x_rank[x]) + + for i in range(n): + df[f'第{str(i+1).zfill(2)}层'] = np.where(df[f'{x_col}分层'] == f'第{str(i+1).zfill(2)}层', df[y_col], 0) + + layering_cols = [f'第{str(i).zfill(2)}层' for i in range(1, n + 1)] + mrr = df[['dt'] + layering_cols].copy() + mrr.set_index('dt', inplace=True) + + tabs = st.tabs(["分层收益率", "多空组合"]) + + with tabs[0]: + show_daily_return(mrr) + + with tabs[1]: + col1, col2 = st.columns(2) + long = col1.multiselect("多头组合", layering_cols, default=["第02层"], key="symbol_factor_long") + short = col2.multiselect("空头组合", layering_cols, default=["第01层"], key="symbol_factor_short") + dfr = mrr.copy() + dfr['多头'] = dfr[long].mean(axis=1) + dfr['空头'] = -dfr[short].mean(axis=1) + dfr['多空'] = (dfr['多头'] + dfr['空头']) / 2 + show_daily_return(dfr[['多头', '空头', '多空']]) From b39f536dfd151648fc52637b0577802444a24898 Mon Sep 17 00:00:00 2001 From: zengbin93 Date: Sat, 14 Oct 2023 18:46:42 +0800 Subject: [PATCH 3/5] 0.9.32 update --- czsc/utils/st_components.py | 1 + 1 file changed, 1 insertion(+) diff --git a/czsc/utils/st_components.py b/czsc/utils/st_components.py index 2d8316bb6..beb5f4787 100644 --- a/czsc/utils/st_components.py +++ b/czsc/utils/st_components.py @@ -185,6 +185,7 @@ def show_symbol_factor_layering(df, x_col, y_col='n1b', **kwargs): if df[x_col].nunique() > n: df[f'{x_col}分层'] = pd.qcut(df[x_col], q=n, labels=False, duplicates='drop') + df[f'{x_col}分层'] = df[f'{x_col}分层'].apply(lambda x: f'第{str(x+1).zfill(2)}层') else: # 如果因子值的取值数量小于分层数量，直接使用因子独立值排序作为分层 x_rank = sorted(df[x_col].unique()) From 24c1d85adf819465c43100f78a6db51b1c728a98 Mon Sep 17 00:00:00 2001 From: zengbin93 Date: Sat, 14 Oct 2023 22:41:47 +0800 Subject: [PATCH 4/5] =?UTF-8?q?0.9.32=20=E4=BC=98=E5=8C=96=20RedisWeightsC?= =?UTF-8?q?lient?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- czsc/traders/rwc.py | 50 ++++++++++++++++++++++++++------------------- 1 file changed, 29 insertions(+), 21 deletions(-) diff --git a/czsc/traders/rwc.py b/czsc/traders/rwc.py index 05e7a6acc..8a4876bee 100644 --- a/czsc/traders/rwc.py +++ b/czsc/traders/rwc.py @@ -20,7 +20,7 @@ class RedisWeightsClient: """策略持仓权重收发客户端""" - version = "V231012" + version = "V231014" def __init__(self, strategy_name, redis_url, **kwargs): """ @@ -81,22 +81,22 @@ def metadata(self): key = f'{self.key_prefix}:META:{self.strategy_name}' return self.r.hgetall(key) - @property - def last_time(self): - """获取策略最近一次发布信号的时间""" - keys = self.get_keys(f'{self.key_prefix}:{self.strategy_name}:*:LAST') - dt = None - for key in keys: - dt_ = pd.to_datetime(str(self.r.hget(key, 'dt'))) - dt = dt_ if not dt else max(dt, dt_) - return dt - - def get_last_time(self, symbol): - """获取指定品种上策略最近一次发布信号的时间""" - key = f'{self.key_prefix}:{self.strategy_name}:{symbol}:LAST' - if not self.r.exists(key): - return None - return pd.to_datetime(str(self.r.hget(key, 'dt'))) + def get_last_times(self, symbols=None): + """获取所有品种上策略最近一次发布信号的时间 + + :param symbols: list, 品种列表, 默认为None, 即获取所有品种 + :return: dict, {symbol: datetime}，如{'SFIF9001': datetime(2021, 9, 24, 15, 19, 0)} + """ + if isinstance(symbols, str): + row = self.r.hgetall(f'{self.key_prefix}:{self.strategy_name}:{symbols}:LAST') + return pd.to_datetime(row['dt']) if row else None # type: ignore + + symbols = symbols if symbols else self.get_symbols() + with self.r.pipeline() as pipe: + for symbol in symbols: + pipe.hgetall(f'{self.key_prefix}:{self.strategy_name}:{symbol}:LAST') + rows = pipe.execute() + return {x['symbol']: pd.to_datetime(x['dt']) for x in rows} def publish(self, symbol, dt, weight, price=0, ref=None, overwrite=False): """发布单个策略持仓权重 @@ -113,7 +113,7 @@ def publish(self, symbol, dt, weight, price=0, ref=None, overwrite=False): dt = pd.to_datetime(dt) if not overwrite: - last_dt = self.get_last_time(symbol) + last_dt = self.get_last_times(symbol) if last_dt is not None and dt <= last_dt: logger.warning(f"不允许重复写入，已过滤 {symbol} {dt} 的重复信号") return 0 @@ -143,9 +143,10 @@ def publish_dataframe(self, df, overwrite=False, batch_size=10000): if not overwrite: raw_count = len(df) + _time = self.get_last_times() _data = [] for symbol, dfg in df.groupby('symbol'): - last_dt = self.get_last_time(symbol) + last_dt = _time.get(symbol) if last_dt is not None: dfg = dfg[dfg['dt'] > last_dt] _data.append(dfg) @@ -250,8 +251,13 @@ def get_symbols(self): symbols = {x.split(":")[2] for x in keys} return list(symbols) - def get_last_weights(self, symbols=None): - """获取最近的持仓权重""" + def get_last_weights(self, symbols=None, ignore_zero=True): + """获取最近的持仓权重 + + :param symbols: list, 品种列表 + :param ignore_zero: boolean, 是否忽略权重为0的品种 + :return: pd.DataFrame + """ symbols = symbols if symbols else self.get_symbols() with self.r.pipeline() as pipe: for symbol in symbols: @@ -261,6 +267,8 @@ def get_last_weights(self, symbols=None): dfw = pd.DataFrame(rows) dfw['weight'] = dfw['weight'].astype(float) dfw['dt'] = pd.to_datetime(dfw['dt']) + if ignore_zero: + dfw = dfw[dfw['weight'] != 0].copy().reset_index(drop=True) return dfw def get_hist_weights(self, symbol, sdt, edt) -> pd.DataFrame: From d3e2a975d00457d9459b5fdd5fc0af3f06c55772 Mon Sep 17 00:00:00 2001 From: zengbin93 Date: Mon, 16 Oct 2023 12:00:10 +0800 Subject: [PATCH 5/5] =?UTF-8?q?0.9.32=20=E4=BC=98=E5=8C=96QMT=E5=AF=B9?= =?UTF-8?q?=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- czsc/connectors/qmt_connector.py | 199 ++++++++++++++++++++++++++++++- 1 file changed, 198 insertions(+), 1 deletion(-) diff --git a/czsc/connectors/qmt_connector.py b/czsc/connectors/qmt_connector.py index ce26309e2..582398cc6 100644 --- a/czsc/connectors/qmt_connector.py +++ b/czsc/connectors/qmt_connector.py @@ -13,7 +13,6 @@ 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 @@ -374,6 +373,204 @@ def on_account_status(self, status): self.push_message(msg, msg_type='text') +def query_stock_positions(xtt: XtQuantTrader, acc: StockAccount): + """查询股票市场的持仓单 + + http://docs.thinktrader.net/pages/ee0e9b/#%E6%8C%81%E4%BB%93%E6%9F%A5%E8%AF%A2 + http://docs.thinktrader.net/pages/198696/#%E6%8C%81%E4%BB%93xtposition + """ + res = xtt.query_stock_positions(acc) + res = {x.stock_code: x for x in res} if len(res) > 0 else {} + return res + + +def query_today_trades(xtt: XtQuantTrader, acc: StockAccount): + """查询当日成交 + + http://docs.thinktrader.net/pages/198696/#%E6%88%90%E4%BA%A4xttrade + """ + trades = xtt.query_stock_trades(acc) + res = [{'品种': x.stock_code, '均价': x.traded_price, "方向": "买入" if x.order_type == 23 else "卖出", + '数量': x.traded_volume, '金额': x.traded_amount, + '时间': time.strftime("%H:%M:%S", time.localtime(x.traded_time))} for x in trades] + return res + + +def cancel_timeout_orders(xtt: XtQuantTrader, acc: StockAccount, minutes=30): + """撤销超时的委托单 + + http://docs.thinktrader.net/pages/ee0e9b/#%E8%82%A1%E7%A5%A8%E5%90%8C%E6%AD%A5%E6%92%A4%E5%8D%95 + http://docs.thinktrader.net/pages/ee0e9b/#%E5%A7%94%E6%89%98%E6%9F%A5%E8%AF%A2 + http://docs.thinktrader.net/pages/198696/#%E5%A7%94%E6%89%98xtorder + + :param minutes: 超时时间，单位分钟 + :return: + """ + orders = xtt.query_stock_orders(acc, cancelable_only=True) + for o in orders: + if datetime.fromtimestamp(o.order_time) < datetime.now() - timedelta(minutes=minutes): + xtt.cancel_order_stock(acc, o.order_id) + + +def is_order_exist(xtt: XtQuantTrader, acc: StockAccount, symbol, order_type, volume=None): + """判断是否存在相同的委托单 + + http://docs.thinktrader.net/pages/ee0e9b/#%E8%82%A1%E7%A5%A8%E5%90%8C%E6%AD%A5%E6%92%A4%E5%8D%95 + http://docs.thinktrader.net/pages/ee0e9b/#%E5%A7%94%E6%89%98%E6%9F%A5%E8%AF%A2 + http://docs.thinktrader.net/pages/198696/#%E5%A7%94%E6%89%98xtorder + + """ + orders = xtt.query_stock_orders(acc, cancelable_only=False) + 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(xtt: XtQuantTrader, acc: StockAccount, symbol, price, **kwargs): + """判断是否允许开仓 + + http://docs.thinktrader.net/pages/198696/#%E8%B5%84%E4%BA%A7xtasset + + :param symbol: 股票代码 + :param price: 股票现价 + :return: True 允许开仓，False 不允许开仓 + """ + symbol_max_pos = kwargs.get('max_pos', 0) # 最大持仓数量 + + # 如果 symbol_max_pos 为 0，不允许开仓 + if symbol_max_pos <= 0: + return False + + # 如果 symbol 在禁止交易的列表中，不允许开仓 + if symbol in kwargs.get('forbidden_symbols', []): + return False + + # 如果 未成交的开仓委托单 存在，不允许开仓 + if is_order_exist(xtt, acc, symbol, order_type=23): + logger.warning(f"存在未成交的开仓委托单，symbol={symbol}") + return False + + # 如果已经有持仓，不允许开仓 + if query_stock_positions(xtt, acc).get(symbol, None): + return False + + # 如果资金不足，不允许开仓 + assets = xtt.query_stock_asset(acc) + if assets.cash < price * 120: + logger.warning(f"资金不足，无法开仓，symbol={symbol}") + return False + + return True + + +def is_allow_exit(xtt: XtQuantTrader, acc: StockAccount, symbol, **kwargs): + """判断是否允许平仓 + + :param symbol: 股票代码 + :return: True 允许开仓，False 不允许开仓 + """ + # symbol 在禁止交易的列表中，不允许平仓 + if symbol in kwargs.get('forbidden_symbols', []): + return False + + # 没有持仓 或 可用数量为 0，不允许平仓 + pos = query_stock_positions(xtt, acc).get(symbol, None) + if not pos or pos.can_use_volume <= 0: + return False + + # 未成交的平仓委托单 存在，不允许继续平仓 + if is_order_exist(xtt, acc, symbol, order_type=24): + logger.warning(f"存在未成交的平仓委托单，symbol={symbol}") + return False + + return True + + +def send_stock_order(xtt: XtQuantTrader, acc: StockAccount, **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 + 委托类型(order_type) http://docs.thinktrader.net/pages/198696/#%E5%A7%94%E6%89%98%E7%B1%BB%E5%9E%8B-order-type + 报价类型(price_type) 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: 委托数量, 股票以'股'为单位, 债券以'张'为单位，ETF以'份'为单位；数量必须是100的整数倍 + 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', 5) + price = kwargs.get('price', 0) + strategy_name = kwargs.get('strategy_name', "程序下单") + order_remark = kwargs.get('order_remark', "程序下单") + + if not xtt.connected: + xtt.start() + xtt.connect() + + order_volume = max(order_volume // 100 * 100, 0) # 股票市场只允许做多 100 的整数倍 + assert xtt.connected, "交易服务器连接断开" + _id = xtt.order_stock(acc, stock_code, order_type, int(order_volume), price_type, price, strategy_name, order_remark) + return _id + + +def order_stock_target(xtt: XtQuantTrader, acc: StockAccount, symbol, target, **kwargs): + """下单调整至目标仓位 + + :param xtt: XtQuantTrader, QMT 交易接口 + :param acc: StockAccount, 账户 + :param symbol: str, 股票代码 + :param target: int, 目标仓位, 单位：股；正数表示持有多头仓位，负数表示持有空头仓位 + :param kwargs: dict, 其他参数 + + - price_type: int, 报价类型, 详见帮助手册 + xtconstant.LATEST_PRICE 5 最新价 + xtconstant.FIX_PRICE 11 限价 + - price: float, 报价价格, 如果price_type为限价, 那price为指定的价格, 否则填0 + + :return: + """ + # 查询持仓 + pos = query_stock_positions(xtt, acc).get(symbol, None) + current = pos.volume if pos else 0 + + logger.info(f"当前持仓：{current}，目标仓位：{target}") + if current == target: + return + + price_type = kwargs.get('price_type', 5) + price = kwargs.get('price', 0) + + # 如果目标小于当前，平仓 + if target < current: + delta = min(current - target, pos.can_use_volume if pos else current) + logger.info(f"{symbol}平仓，目标仓位：{target}，当前仓位：{current}，平仓数量：{delta}") + if delta != 0: + send_stock_order(xtt, acc, stock_code=symbol, order_type=24, + order_volume=delta, price_type=price_type, price=price) + return + + # 如果目标大于当前，开仓 + if target > current: + delta = target - current + logger.info(f"{symbol}开仓，目标仓位：{target}，当前仓位：{current}，开仓数量：{delta}") + if delta != 0: + send_stock_order(xtt, acc, stock_code=symbol, order_type=23, + order_volume=delta, price_type=price_type, price=price) + return + + class QmtTradeManager: """QMT交易管理器（这是一个案例性质的存在，真正实盘的时候请参考这个，根据自己的逻辑重新实现）