124 lines
4.1 KiB
Python
124 lines
4.1 KiB
Python
import json
|
|
from typing import Dict, TypedDict
|
|
import sys
|
|
import os
|
|
sys.path.append(os.path.abspath("/Users/jack/Project/chan.py"))
|
|
import xgboost as xgb
|
|
from BuySellPoint.BS_Point import CBS_Point
|
|
from Chan import CChan
|
|
from ChanConfig import CChanConfig
|
|
from ChanModel.Features import CFeatures
|
|
from Common.CEnum import AUTYPE, DATA_SRC, KL_TYPE, BSP_TYPE, MACD_ALGO
|
|
from Common.CTime import CTime
|
|
from Plot.PlotDriver import CPlotDriver
|
|
from Bi.Bi import CBi
|
|
|
|
class FeatureDict:
|
|
def klc_features(self, klc):
|
|
# klc features
|
|
klc_feature_dict = {
|
|
"klu_max_high": klc.get_klu_max_high(),
|
|
"klu_max_low": klc.get_klu_min_low(),
|
|
|
|
}
|
|
return klc_feature_dict
|
|
def bi_features(self, bi):
|
|
# bi features
|
|
bi_feature_dict = {
|
|
"bi_macd_area": bi.cal_macd_metric(MACD_ALGO.AREA, False),
|
|
"bi_macd_peak": bi.cal_macd_metric(MACD_ALGO.PEAK, False),
|
|
"bi_macd_full_area": bi.cal_macd_metric(MACD_ALGO.FULL_AREA, False),
|
|
"bi_macd_diff": bi.cal_macd_metric(MACD_ALGO.DIFF, False),
|
|
"bi_macd_slop": bi.cal_macd_metric(MACD_ALGO.SLOPE, False),
|
|
"bi_macd_amp": bi.cal_macd_metric(MACD_ALGO.AMP, False),
|
|
"bi_macd_amount": bi.cal_macd_metric(MACD_ALGO.AMOUNT, False),
|
|
"bi_macd_rsi": bi.cal_macd_metric(MACD_ALGO.RSI, False),
|
|
}
|
|
#bi_feature_dict.update(self.klc_features(bi.begin_klc))
|
|
#bi_feature_dict.update(self.klc_features(bi.end_klc))
|
|
return bi_feature_dict
|
|
|
|
def seg_features(self, seg):
|
|
seg_feature_dict = {
|
|
# seg features
|
|
"seg_macd_slope": seg.Cal_MACD_slope(),
|
|
"seg_macd_amp": seg.Cal_MACD_amp(),
|
|
"seg_amp": seg.amp(),
|
|
"seg_cal_klu_slope": seg.cal_klu_slope(),
|
|
"seg_cal_amp": seg.amp(),
|
|
}
|
|
return seg_feature_dict
|
|
|
|
def zs_features(self, zs):
|
|
zs_feature_dict = {
|
|
f"zs_high": zs.high,
|
|
f"zs_low": zs.low,
|
|
f"zs_mid": zs.mid,
|
|
f"zs_peak_low": zs.peak_low,
|
|
f"zs_peak_high": zs.peak_high,
|
|
}
|
|
#zs_feature_dict.update(bi_features(zs.begin_bi))
|
|
#zs_feature_dict.update(bi_features(zs.end_bi))
|
|
#zs_feature_dict.update(bi_features(zs.bi_in))
|
|
#zs_feature_dict.update(bi_features(zs.bi_out))
|
|
return zs_feature_dict
|
|
|
|
def klu_features(self, klu):
|
|
klu_feature_dict = {
|
|
# klu features
|
|
"klu_rsi": klu.rsi,
|
|
"klu_macd_fast_ema": klu.macd.fast_ema,
|
|
"klu_macd_slow_ema": klu.macd.slow_ema,
|
|
"klu_macd_DIF": klu.macd.DIF,
|
|
"klu_macd_DEA": klu.macd.DEA,
|
|
"klu_macd_macd": klu.macd.macd,
|
|
"klu_kdj_k": klu.kdj.k,
|
|
"klu_kdj_d": klu.kdj.d,
|
|
"klu_kdj_j": klu.kdj.j,
|
|
}
|
|
return klu_feature_dict
|
|
|
|
def stragety_feature(self, last_bsp):
|
|
last_bi = last_bsp.bi
|
|
last_klu = last_bsp.klu
|
|
feature_dict = {
|
|
"open_klu_rate": (last_klu.close - last_klu.open)/last_klu.open,
|
|
|
|
}
|
|
# add klu features
|
|
feature_dict.update(self.klu_features(last_klu))
|
|
# add bi features
|
|
#feature_dict.update(self.bi_features(last_bi))
|
|
# add seg features
|
|
seg = last_bi.parent_seg
|
|
#feature_dict.update(seg_features(seg))
|
|
|
|
# add zs features
|
|
zs_list = seg.zs_lst
|
|
zs_feature_dict = {}
|
|
if zs_list:
|
|
zs_feature_dict = {
|
|
f"zs_count": len(zs_list),
|
|
}
|
|
bi_in_zs = False
|
|
for zs in zs_list:
|
|
if last_bi in zs.bi_lst:
|
|
bi_in_zs = True
|
|
if bi_in_zs:
|
|
zs_feature_dict.update({
|
|
"bi_in_zs": 1
|
|
})
|
|
else:
|
|
zs_feature_dict.update({
|
|
"bi_in_zs": 0
|
|
})
|
|
zs_feature_dict.update(self.zs_features(zs_list[-1]))
|
|
else:
|
|
zs_feature_dict ={
|
|
"zs_count": 0,
|
|
}
|
|
feature_dict.update(zs_feature_dict)
|
|
return feature_dict
|
|
|
|
|