0.5.2 修复bug，辅助技术指标计算

czsc/analyze.py

@@ -161,12 +161,15 @@ class KlineAnalyze:
             ma_ = {'ma%i' % p: sum([x['close'] for x in self.kline_raw[-p:]]) / p
                    for p in self.ma_params}
             ma_.update({"dt": self.kline_raw[-1]['dt']})
+            if self.verbose:
+                print("ma new: %s" % str(ma_))
+
             if self.kline_raw[-2]['dt'] == self.ma[-1]['dt']:
                 self.ma.append(ma_)
             else:
                 self.ma[-1] = ma_
 
-        assert len(self.ma) == len(self.kline_raw)
+        assert self.ma[-2]['dt'] == self.kline_raw[-2]['dt']
 
         if not self.macd:
             close_ = np.array([x["close"] for x in self.kline_raw], dtype=np.double)
@@ -189,11 +192,16 @@ class KlineAnalyze:
                     "dea": m2[-1],
                     "macd": m3[-1]
                 }
-            if self.kline_raw[-2]['dt'] == self.ma[-1]['dt']:
+            if self.verbose:
+                print("macd new: %s" % str(macd_))
+
+            if self.kline_raw[-2]['dt'] == self.macd[-1]['dt']:
                 self.macd.append(macd_)
             else:
                 self.macd[-1] = macd_
 
+        assert self.macd[-2]['dt'] == self.kline_raw[-2]['dt']
+
     def _update_kline_new(self):
         """更新去除包含关系的K线序列
 
@@ -534,6 +542,8 @@ class KlineAnalyze:
         # 根据最大原始K线序列长度限制分析结果长度
         if len(self.kline_raw) > self.max_raw_len:
             self.kline_raw = self.kline_raw[-self.max_raw_len:]
+            self.ma = self.ma[-self.max_raw_len:]
+            self.macd = self.macd[-self.max_raw_len:]
             self.kline_new = self.kline_new[-self.max_raw_len:]
             self.fx_list = self.fx_list[-(self.max_raw_len//2):]
             self.bi_list = self.bi_list[-(self.max_raw_len//4):]

test/test_analyze.py

@@ -13,13 +13,19 @@ from czsc.analyze import KlineAnalyze, find_zs
 
 warnings.warn(f"czsc version is {czsc.__version__}")
 
-# cur_path = os.path.split(os.path.realpath(__file__))[0]
-cur_path = "./test"
+cur_path = os.path.split(os.path.realpath(__file__))[0]
+# cur_path = "./test"
 file_kline = os.path.join(cur_path, "data/000001.SH_D.csv")
 kline = pd.read_csv(file_kline, encoding="utf-8")
 kline.loc[:, "dt"] = pd.to_datetime(kline.dt)
-ka = KlineAnalyze(kline, name="日线", max_raw_len=2000)
-
+kline1 = kline.iloc[:2000]
+kline2 = kline.iloc[2000:]
+ka = KlineAnalyze(kline1, name="日线", max_raw_len=2000, verbose=True)
+
+def test_update():
+    for _, row in kline2.iterrows():
+        ka.update(row.to_dict())
+        assert ka.kline_raw[-1]['dt'] == row['dt']
 
 def test_update_ta():
     ma_x1 = dict(ka.ma[-1])