0.9.30 update

czsc/__init__.py

@@ -42,6 +42,10 @@
     BarGenerator,
     freq_end_time,
     resample_bars,
+    is_trading_time,
+    get_intraday_times,
+    check_freq_and_market,
+
     dill_dump,
     dill_load,
     read_json,

czsc/traders/weight_backtest.py

@@ -321,12 +321,6 @@ def report(self, res_path):
 
         # 品种等权费后日收益率
         dret = res['品种等权日收益'].copy()
-        # dret = pd.concat([v['daily'] for v in res.values()], ignore_index=True)
-        # dret = pd.pivot_table(dret, index='date', columns='symbol', values='return').fillna(0)
-        # dret['total'] = dret[list(res.keys())].mean(axis=1)
-        # stats = {"开始日期": dret.index.min().strftime("%Y%m%d"), "结束日期": dret.index.max().strftime("%Y%m%d")}
-        # stats.update(daily_performance(dret['total']))
-        # logger.info(f"品种等权费后日收益率：{stats}")
         dret.to_excel(res_path.joinpath("daily_return.xlsx"), index=True)
         logger.info(f"品种等权费后日收益率已保存到 {res_path.joinpath('daily_return.xlsx')}")
 
@@ -339,11 +333,6 @@ def report(self, res_path):
 
         # 所有开平交易记录的表现
         stats = res['绩效评价'].copy()
-        # dfp = pd.concat([v['pairs'] for v in res.values()], ignore_index=True)
-        # pairs_stats = evaluate_pairs(dfp)
-        # pairs_stats = {k: v for k, v in pairs_stats.items() if k in ['单笔收益', '持仓K线数', '交易胜率', '持仓天数']}
-        # logger.info(f"所有开平交易记录的表现：{pairs_stats}")
-        # stats.update(pairs_stats)
         logger.info(f"绩效评价：{stats}")
         save_json(stats, res_path.joinpath("stats.json"))
         logger.info(f"绩效评价已保存到 {res_path.joinpath('stats.json')}")

czsc/utils/__init__.py

@@ -11,6 +11,7 @@
 from .word_writer import WordWriter
 from .corr import nmi_matrix, single_linear, cross_sectional_ic
 from .bar_generator import BarGenerator, freq_end_time, resample_bars
+from .bar_generator import is_trading_time, get_intraday_times, check_freq_and_market
 from .io import dill_dump, dill_load, read_json, save_json
 from .sig import check_pressure_support, check_gap_info, is_bis_down, is_bis_up, get_sub_elements
 from .sig import same_dir_counts, fast_slow_cross, count_last_same, create_single_signal

czsc/utils/bar_generator.py

@@ -22,6 +22,24 @@
         freq_edt_map[f"{_f}_{_m}"] = {k: v for k, v in dfg[["time", _f]].values}
 
 
+def is_trading_time(dt: datetime = datetime.now(), market="A股"):
+    """判断指定时间是否是交易时间"""
+    hm = dt.strftime("%H:%M")
+    times = freq_market_times[f"1分钟_{market}"]
+    return True if hm in times else False
+
+
+def get_intraday_times(freq='1分钟', market="A股"):
+    """获取指定市场的交易时间段
+
+    :param market: 市场名称，可选值：A股、期货、默认
+    :return: 交易时间段列表
+    """
+    assert market in ['A股', '期货', '默认'], "market 参数必须为 A股 或 期货 或 默认"
+    assert freq.endswith("分钟"), "freq 参数必须为分钟级别的K线周期"
+    return freq_market_times[f"{freq}_{market}"]
+
+
 def check_freq_and_market(time_seq: List[AnyStr]):
     """检查时间序列是否为同一周期，是否为同一市场
 

test/test_bar_generator.py

@@ -338,3 +338,33 @@ def test_bg_on_d():
     assert len(bg.bars['月线']) == 165
     assert len(bg.bars['年线']) == 15
     assert bg.bars['月线'][-2].id > bg.bars['月线'][-3].id
+
+
+def test_is_trading_time():
+    from datetime import datetime
+    from czsc.utils.bar_generator import is_trading_time
+
+    # Test for A股 market
+    assert not is_trading_time(datetime(2022, 1, 3, 9, 30), market="A股")
+    assert is_trading_time(datetime(2022, 1, 3, 9, 31), market="A股")
+    assert is_trading_time(datetime(2022, 1, 3, 11, 30), market="A股")
+    assert not is_trading_time(datetime(2022, 1, 3, 12, 59), market="A股")
+    assert is_trading_time(datetime(2022, 1, 3, 15, 0), market="A股")
+    assert not is_trading_time(datetime(2022, 1, 3, 20, 0), market="A股")
+
+    # Test for other markets
+    assert is_trading_time(datetime(2022, 1, 3, 9, 30), market="期货")
+    assert not is_trading_time(datetime(2022, 1, 3, 10, 25), market="期货")
+    assert not is_trading_time(datetime(2022, 1, 3, 12, 59), market="期货")
+    assert is_trading_time(datetime(2022, 1, 3, 15, 0), market="期货")
+    assert not is_trading_time(datetime(2022, 1, 3, 20, 0), market="期货")
+
+
+def test_get_intraday_times():
+    from czsc.utils.bar_generator import get_intraday_times
+
+    assert get_intraday_times(freq='60分钟', market='A股') == ['10:30', '11:30', '14:00', '15:00']
+    assert get_intraday_times(freq='120分钟', market='A股') == ['11:30', '15:00']
+    assert get_intraday_times(freq='120分钟', market='期货') == ['11:00', '15:00', '23:00', '01:00', '02:30']
+    x = ['02:00', '04:00', '06:00', '08:00', '10:00', '12:00', '14:00', '16:00', '18:00', '20:00', '22:00', '00:00']
+    assert get_intraday_times(freq='120分钟', market='默认') == x