V0.9.28 更新一批代码

.github/workflows/pythonpackage.yml

@@ -5,7 +5,7 @@ name: Python package
 
 on:
   push:
-    branches: [ master, V0.9.27 ]
+    branches: [ master, V0.9.28 ]
   pull_request:
     branches: [ master ]
 

README.md

@@ -17,8 +17,9 @@
 
 >**假如没有了分型、笔、线段，缠论还是缠论吗？如果你的答案是“是”，这个项目是为你准备的。本项目旨在提供一个符合缠中说禅思维方式的程序化交易工具。**
 
-* 已经开始用czsc库进行量化研究的朋友，欢迎加入飞书群，快点击 https://applink.feishu.cn/client/chat/chatter/add_by_link?link_token=a51qf1f4-8aee-47a5-8dca-e01d076b4baf 加入吧！
+* 已经开始用czsc库进行量化研究的朋友，欢迎[加入飞书群](https://applink.feishu.cn/client/chat/chatter/add_by_link?link_token=a51qf1f4-8aee-47a5-8dca-e01d076b4baf)，快点击加入吧！
 * [B站视频教程合集（持续更新...）](https://space.bilibili.com/243682308/channel/series)
+* [CZSC策略圈介绍](https://s0cqcxuy3p.feishu.cn/wiki/D12bwh4SriW1Lgk23HUchFKFnpe)
 
 
 ## 项目贡献
@@ -91,15 +92,3 @@ pip install czsc -U -i https://pypi.python.org/simple
 
 * 链接：https://pan.baidu.com/s/1RXkP3188F0qu8Yk6CjbxRQ
 * 提取码：vhue
-
-
-## [知识星球【CZSC小圈子】](https://t.zsxq.com/04B2jmUN7)费用：100元
-
-**知识星球【CZSC小圈子】的定位是什么？**
-  - 为仔细研读过禅师原文并且愿意使用 CZSC 库进行量化投研的朋友提供一个交流平台。
-  - 寻找一群有能力的朋友共同进行量化策略研究。
-  - 促成策略逻辑互补的实盘组合构建。
-  - 对于刚接触缠论和量化交易的新朋友，给出一些力所能及的帮助。
-> 详情点击：https://s0cqcxuy3p.feishu.cn/wiki/wikcnwXSk9mWnki1b6URPhLA2Hc
-> 
-> **加入知识星球后**，请加微信 `zengbin93`，备注【CZSC小圈子】

czsc/__init__.py

@@ -12,37 +12,76 @@
 from czsc import aphorism
 from czsc.analyze import CZSC
 from czsc.objects import Freq, Operate, Direction, Signal, Factor, Event, RawBar, NewBar, Position
-
-from czsc.traders import CzscTrader, CzscSignals, generate_czsc_signals, check_signals_acc, get_unique_signals
-from czsc.traders import PairsPerformance, combine_holds_and_pairs, combine_dates_and_pairs, stock_holds_performance
-from czsc.traders import DummyBacktest, SignalsParser, get_signals_by_conf, get_signals_config, get_signals_freqs
-from czsc.traders import WeightBacktest, get_ensemble_weight
-from czsc.sensors import holds_concepts_effect, CTAResearch, EventMatchSensor
 from czsc.strategies import CzscStrategyBase, CzscJsonStrategy
-
-from czsc.utils import KlineChart, BarGenerator, resample_bars, dill_dump, dill_load, read_json, save_json
-from czsc.utils import get_sub_elements, get_py_namespace, freqs_sorted, x_round, import_by_name, create_grid_params
-from czsc.utils import cal_trade_price, cross_sectional_ic, update_bbars, update_tbars, update_nbars
-from czsc.utils import CrossSectionalPerformance
-from czsc.utils.signal_analyzer import SignalAnalyzer, SignalPerformance
-from czsc.utils.stats import daily_performance, net_value_stats, subtract_fee
-from czsc.utils.cache import home_path, get_dir_size, empty_cache_path
+from czsc.sensors import holds_concepts_effect, CTAResearch, EventMatchSensor
+from czsc.traders import (
+    CzscTrader,
+    CzscSignals,
+    generate_czsc_signals,
+    check_signals_acc,
+    get_unique_signals,
+    PairsPerformance,
+    combine_holds_and_pairs,
+    combine_dates_and_pairs,
+    stock_holds_performance,
+    DummyBacktest,
+    SignalsParser,
+    get_signals_by_conf,
+    get_signals_config,
+    get_signals_freqs,
+    WeightBacktest,
+    get_ensemble_weight,
+)
+from czsc.utils import (
+    KlineChart,
+    WordWriter,
+    BarGenerator,
+    freq_end_time,
+    resample_bars,
+    dill_dump,
+    dill_load,
+    read_json,
+    save_json,
+    get_sub_elements,
+    get_py_namespace,
+    freqs_sorted,
+    x_round,
+    import_by_name,
+    create_grid_params,
+    cal_trade_price,
+    update_bbars,
+    update_tbars,
+    update_nbars,
+    CrossSectionalPerformance,
+    cross_sectional_ranker,
+    cross_sectional_ic,
+    SignalAnalyzer,
+    SignalPerformance,
+    daily_performance,
+    net_value_stats,
+    subtract_fee,
+    home_path,
+    get_dir_size,
+    empty_cache_path,
+)
 
 
-__version__ = "0.9.27"
+__version__ = "0.9.28"
 __author__ = "zengbin93"
 __email__ = "zeng_bin8888@163.com"
-__date__ = "20230812"
+__date__ = "20230820"
 
 
 def welcome():
     print(f"欢迎使用CZSC！当前版本标识为 {__version__}@{__date__}\n")
     aphorism.print_one()
 
-    print(f"CZSC环境变量："
-          f"czsc_min_bi_len = {envs.get_min_bi_len()}; "
-          f"czsc_max_bi_num = {envs.get_max_bi_num()}; "
-          f"czsc_bi_change_th = {envs.get_bi_change_th()}")
+    print(
+        f"CZSC环境变量："
+        f"czsc_min_bi_len = {envs.get_min_bi_len()}; "
+        f"czsc_max_bi_num = {envs.get_max_bi_num()}; "
+        f"czsc_bi_change_th = {envs.get_bi_change_th()}"
+    )
 
 
 if envs.get_welcome():

czsc/analyze.py

@@ -7,9 +7,8 @@
 """
 import os
 import webbrowser
-import numpy as np
 from loguru import logger
-from typing import List, Callable
+from typing import List
 from collections import OrderedDict
 from czsc.enum import Mark, Direction
 from czsc.objects import BI, FX, RawBar, NewBar

czsc/data/ts.py

@@ -93,10 +93,10 @@ def format_kline(kline: pd.DataFrame, freq: Freq) -> List[RawBar]:
 
     for i, record in enumerate(records):
         if freq == Freq.D:
-            vol = int(record['vol']*100)
-            amount = int(record.get('amount', 0)*1000)
+            vol = int(record['vol'] * 100) if record['vol'] > 0 else 0
+            amount = int(record.get('amount', 0) * 1000)
         else:
-            vol = int(record['vol'])
+            vol = int(record['vol']) if record['vol'] > 0 else 0
             amount = int(record.get('amount', 0))
 
         # 将每一根K线转换成 RawBar 对象

czsc/sensors/event.py

@@ -6,7 +6,7 @@
 describe: Event 相关的传感器
 """
 import os
-import shutil   
+import shutil
 import pandas as pd
 from copy import deepcopy
 from loguru import logger
@@ -83,7 +83,7 @@ def __init__(self, events: List[Union[Dict[str, Any], Event]], symbols: List[str
             df = df.set_index("dt")
             _res.append(df)
         df = pd.concat(_res, axis=1, ignore_index=False).reset_index()
-        file_csc = os.path.join(self.results_path, f"cross_section_counts.csv")
+        file_csc = os.path.join(self.results_path, "cross_section_counts.csv")
         df.to_csv(file_csc, index=False)
         logger.info(f"截面匹配次数计算完成，结果保存至：{file_csc}")
 

czsc/signals/__init__.py

@@ -44,6 +44,7 @@
     cxt_ubi_end_V230816,
     cxt_bi_end_V230815,
     cxt_bi_stop_V230815,
+    cxt_bi_trend_V230824,
 )
 
 
@@ -232,6 +233,8 @@
     obvm_line_V230610,
     obv_up_dw_line_V230719,
     cvolp_up_dw_line_V230612,
+    kcatr_up_dw_line_V230823,
+    ntmdk_V230824,
 )
 
 

czsc/signals/ang.py

@@ -794,3 +794,96 @@ def cvolp_up_dw_line_V230612(c: CZSC, **kwargs) -> OrderedDict:
         v1 = "看空"
 
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def ntmdk_V230824(c: CZSC, **kwargs) -> OrderedDict:
+    """NTMDK多空指标，贡献者：琅盎  
+
+    参数模板："{freq}_D{di}M{m}_NTMDK多空V230824"
+
+    **信号逻辑：**
+
+    此信号函数的逻辑非常简单，流传于股市中有一句话：日日新高日日持股，
+    那么此信号函数利用的是收盘价和M日前的收盘价进行比较，如果差值为正
+    即多头成立，反之空头成立。
+
+    **信号列表：**
+
+    - Signal('日线_D1M10_NTMDK多空V230824_看空_任意_任意_0')
+    - Signal('日线_D1M10_NTMDK多空V230824_看多_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 参数字典
+
+        - :param di: 信号计算截止倒数第i根K线
+        - :param m: m天前的价格
+
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get("di", 1))
+    m = int(kwargs.get("m", 10))
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}M{m}_NTMDK多空V230824".split('_')
+    v1 = "其他"
+    if len(c.bars_raw) < di + m + 10:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    bars = get_sub_elements(c.bars_raw, di=di, n=m)
+    v1 = "看多" if bars[-1].close > bars[0].close else "看空"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+
+def kcatr_up_dw_line_V230823(c: CZSC, **kwargs) -> OrderedDict:
+    """用ATR波幅构造上下轨，收盘价突破判断多空  贡献者：琅盎
+
+    参数模板："{freq}_D{di}N{n}M{m}T{th}_KCATR多空V230823"
+
+    **信号逻辑：**
+
+    与布林带类似，都是用价格的移动平均构造中轨，不同的是表示波幅
+    的方法，这里用 atr 来作为波幅构造上下轨。价格突破上轨，
+    可看成新的上升趋势，买入；价格突破下轨，
+
+    **信号列表：**
+
+    - Signal('日线_D1N30M16T2_KCATR多空V230823_看多_任意_任意_0')
+    - Signal('日线_D1N30M16T2_KCATR多空V230823_看空_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs: 参数字典
+
+        - :param di: 信号计算截止倒数第i根K线
+        - :param n: 获取K线的根数进行ATR计算，默认为30
+        - :param m: 获取K线的根数进行均价计算，默认为16
+        - :param th: 突破ATR的倍数，默认为2
+
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get("di", 1))
+    n = int(kwargs.get("n", 30))
+    m = int(kwargs.get("m", 16))
+    th = int(kwargs.get("th", 2))  # 突破ATR的倍数
+
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}N{n}M{m}T{th}_KCATR多空V230823".split('_')
+    v1 = "其他"
+    if len(c.bars_raw) < di + max(m, n) + 10:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    n_bars = get_sub_elements(c.bars_raw, di=di, n=n)
+    m_bars = get_sub_elements(c.bars_raw, di=di, n=m)
+    atr = np.mean([
+        max(
+            abs(n_bars[i].high - n_bars[i].low),
+            abs(n_bars[i].high - n_bars[i - 1].close),
+            abs(n_bars[i].low - n_bars[i - 1].close),
+        )
+        for i in range(1, len(n_bars))
+    ])
+    middle = np.mean([x.close for x in m_bars])
+
+    if m_bars[-1].close > middle + atr * th:
+        v1 = "看多"
+    elif m_bars[-1].close < middle - atr * th:
+        v1 = "看空"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)

czsc/signals/cxt.py

@@ -2044,3 +2044,55 @@ def cxt_bi_stop_V230815(c: CZSC, **kwargs) -> OrderedDict:
         v1 = '向上'
         v2 = "阈值内" if last_bar.close < bi.low * (1 + th / 10000) else "阈值外"
     return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1, v2=v2)
+
+
+def cxt_bi_trend_V230824(c: CZSC, **kwargs) -> OrderedDict:
+    """判断N笔形态，贡献者：chenglei
+
+    参数模板："{freq}_D{di}N{n}TH{th}_形态V230824"
+
+    **信号逻辑：**
+
+    1. 通过对最近N笔的中心点的均值和-n笔的中心点的位置关系来判断当前N比是上涨形态还是下跌，横盘震荡形态
+    2. 给定阈值 th，判断上涨下跌横盘按照 所有笔中心点/第-n笔中心点 与 正负th区间的相对位置来判断。
+    3. 当在区间上时为上涨，区间内为横盘，区间下为下跌
+
+    **信号列表：**
+
+    - Signal('日线_D1N4TH5_形态V230824_横盘_任意_任意_0')
+    - Signal('日线_D1N4TH5_形态V230824_向上_任意_任意_0')
+    - Signal('日线_D1N4TH5_形态V230824_向下_任意_任意_0')
+
+    :param c: CZSC对象
+    :param kwargs:
+
+        - di: 倒数第几笔
+        - n ：检查范围
+        - th: 振幅阈值，2 表示 2%，即 2% 以内的振幅都认为是震荡
+
+    :return: 信号识别结果
+    """
+    di = int(kwargs.get('di', 1))
+    n = int(kwargs.get('n', 4))
+    th = int(kwargs.get('th', 2))  # 振幅阈值，2 表示 2%，即 2% 以内的振幅都认为是震荡
+    freq = c.freq.value
+    k1, k2, k3 = f"{freq}_D{di}N{n}TH{th}_形态V230824".split('_')
+    v1 = '其他'
+    if len(c.bi_list) < di + n + 2:
+        return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+
+    _bis = get_sub_elements(c.bi_list, di=di, n=n)
+    assert len(_bis) == n, f"获取第 {di} 笔到第 {di+n} 笔失败"
+
+    all_means = [(bi.low + bi.high) / 2 for bi in _bis]
+    average_of_means = sum(all_means) / n
+    ratio = all_means[0] / average_of_means
+
+    if ratio * 100 > 100 + th:
+        v1 = "向下"
+    elif ratio * 100 < 100 - th:
+        v1 = "向上"
+    else:
+        v1 = "横盘"
+    return create_single_signal(k1=k1, k2=k2, k3=k3, v1=v1)
+

czsc/signals/tas.py

@@ -3322,7 +3322,7 @@ def tas_macd_bc_V230804(c: CZSC, **kwargs) -> OrderedDict:
 def tas_macd_bc_ubi_V230804(c: CZSC, **kwargs) -> OrderedDict:
     """未完成笔MACD黄白线辅助背驰判断
 
-    参数模板："{freq}_MACD背驰_BS辅助V230804"
+    参数模板："{freq}_MACD背驰_UBI观察V230804"
 
     **信号逻辑：**
 

czsc/traders/base.py

@@ -172,7 +172,7 @@ def get_signals_by_conf(cat: CzscSignals, conf):
         sig_func = import_by_name(param.pop('name'))
         freq = param.pop('freq', None)
         if freq in cat.kas:  # 如果指定了 freq，那么就使用 CZSC 对象作为输入
-            s.update(sig_func(cat.kas[freq], **param))
+            s.update(sig_func(cat.kas[freq], **param))   # type: ignore
         else:                # 否则使用 CAT 作为输入
             s.update(sig_func(cat, **param))
     return s
@@ -197,13 +197,13 @@ def generate_czsc_signals(bars: List[RawBar], signals_config: List[dict],
     """
     freqs = get_signals_freqs(signals_config)
     freqs = [freq for freq in freqs if freq != bars[0].freq.value]
-    sdt = pd.to_datetime(sdt) # type: ignore
-    bars_left = [x for x in bars if x.dt < sdt] # type: ignore
+    sdt = pd.to_datetime(sdt)                       # type: ignore
+    bars_left = [x for x in bars if x.dt < sdt]     # type: ignore
     if len(bars_left) <= init_n:
         bars_left = bars[:init_n]
         bars_right = bars[init_n:]
     else:
-        bars_right = [x for x in bars if x.dt >= sdt] # type: ignore
+        bars_right = [x for x in bars if x.dt >= sdt]   # type: ignore
 
     if len(bars_right) == 0:
         logger.warning("右侧K线为空，无法进行信号生成", category=RuntimeWarning)
@@ -302,7 +302,7 @@ def get_unique_signals(bars: List[RawBar], signals_config: List[dict], **kwargs)
 class CzscTrader(CzscSignals):
     """缠中说禅技术分析理论之多级别联立交易决策类（支持多策略独立执行）"""
 
-    def __init__(self, bg: BarGenerator = None, positions: List[Position] = None,
+    def __init__(self, bg: Optional[BarGenerator] = None, positions: Optional[List[Position]] = None,
                  ensemble_method: Union[AnyStr, Callable] = "mean", **kwargs):
         """
 
@@ -483,7 +483,7 @@ def get_ensemble_weight(self, method: Optional[Union[AnyStr, Callable]] = None):
                 {'多头策略A': 1, '多头策略B': 1, '空头策略A': -1}
         :param kwargs:
         :return: pd.DataFrame
-            columns = ['dt', 'symbol', 'weight', 'price']    
+            columns = ['dt', 'symbol', 'weight', 'price']
         """
         from czsc.traders.weight_backtest import get_ensemble_weight
         method = self.__ensemble_method if not method else method