Skip to content

C
czsc

zengbin93 / czsc

通知 23
Star 2
Fork 1
C
czsc

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 503b63e9 编写于 作者: Z zengbin93
浏览文件
操作

0.5.4 首次commit

上级 832debc7
隐藏空白更改
内联 并排
Showing with 1 addition and 183 deletion
czsc/__init__.py
浏览文件 @ 503b63e9
......@@ -3,7 +3,7 @@
from .analyze import KlineAnalyze, find_zs
from .ta import SMA, EMA, MACD
__version__ = "0.5.3"
__version__ = "0.5.4"
__author__ = "zengbin93"
__email__ = "zeng_bin8888@163.com"
......
czsc/signals.py
浏览文件 @ 503b63e9
......@@ -350,188 +350,6 @@ def get_xd_signals(ka, use_zs=False):
return {freq + k: v for k, v in s.items()}
def get_fx_signals_v1(ka):
"""计算分型特征"""
s = {
"SFX01_收于MA5上方": None,
"SFX02_收于MA20上方": None,
"SFX03_收于MA120上方": None,
"SFX04_最近一个分型类型": None,
"SFX05_有效跌破MA5": None,
"SFX06_有效升破MA5": None,
}
last_klines_ = [dict(x) for x in ka.kline_raw[-10:]]
last_ma_ = ka.ma[-10:]
for i, x in enumerate(last_klines_):
assert last_ma_[i]['dt'] == x['dt'], "{}：计算均线错误".format(ka.name)
last_klines_[i].update(last_ma_[i])
last_k = last_klines_[-1]
s["SFX01_收于MA5上方"] = True if last_k['close'] >= last_k['ma5'] > 0 else False
s["SFX02_收于MA20上方"] = True if last_k['close'] >= last_k['ma20'] > 0 else False
s["SFX03_收于MA120上方"] = True if last_k['close'] >= last_k['ma120'] > 0 else False
s["SFX04_最近一个分型类型"] = ka.fx_list[-1]['fx_mark']
if sum([1 for x in last_klines_[-5:] if x['low'] < x['ma5']]) >= 5 \
or sum([1 for x in last_klines_[-3:] if x['close'] < x['ma5']]) >= 3:
s["SFX05_有效跌破MA5"] = True
if sum([1 for x in last_klines_[-5:] if x['high'] > x['ma5']]) >= 5 \
or sum([1 for x in last_klines_[-3:] if x['close'] > x['ma5']]) >= 3:
s["SFX06_有效升破MA5"] = True
freq = ka.name
return {freq + "_" + k: v for k, v in s.items()}
def get_bi_signals_v1(ka):
"""计算笔信号"""
s = {
"SBI01_最近一个未确认的笔标记类型": None,
"SBI02_最近一个已确认的笔标记类型": None,
"SBI03_最近一笔不创新高": False,
"SBI04_最近一笔不创新低": False,
'SBI05_最近一笔进入类盘整顶背驰段': None,
'SBI06_最近一笔进入类盘整底背驰段': None,
'SBI07_最近一笔进入类趋势顶背驰段': None,
'SBI08_最近一笔进入类趋势底背驰段': None,
"SBI09_收盘价相对于最近笔中枢位置": None, # 上面（1）、内部（0）、下面（-1）
"SBI10_最近笔中枢出现三买": False,
"SBI11_最近笔中枢出现三卖": False,
}
# ------------------------------------------------------------------------------------------------------------------
if len(ka.bi_list) > 3:
s["SBI01_最近一个未确认的笔标记类型"] = ka.bi_list[-1]['fx_mark']
s["SBI02_最近一个已确认的笔标记类型"] = ka.bi_list[-2]['fx_mark']
if ka.bi_list[-1]['fx_mark'] == 'g' and ka.bi_list[-1]['bi'] < ka.bi_list[-3]['bi']:
s["SBI03_最近一笔不创新高"] = True
if ka.bi_list[-1]['fx_mark'] == 'd' and ka.bi_list[-1]['bi'] > ka.bi_list[-3]['bi']:
s["SBI04_最近一笔不创新低"] = True
# ------------------------------------------------------------------------------------------------------------------
bis = ka.bi_list[-30:]
if len(bis) >= 6:
if bis[-1]['fx_mark'] == 'd' and bis[-1]['bi'] < bis[-3]['bi'] and bis[-2]['bi'] < bis[-4]['bi']:
zs1 = {"start_dt": bis[-2]['dt'], "end_dt": bis[-1]['dt'], "direction": "down"}
zs2 = {"start_dt": bis[-4]['dt'], "end_dt": bis[-3]['dt'], "direction": "down"}
if ka.is_bei_chi(zs1, zs2, mode="bi", adjust=0.9):
# 类趋势
if bis[-2]['bi'] < bis[-5]['bi']:
s['SBI08_最近一笔进入类趋势底背驰段'] = True
else:
s['SBI06_最近一笔进入类盘整底背驰段'] = True
if bis[-1]['fx_mark'] == 'g' and bis[-1]['bi'] > bis[-3]['bi'] and bis[-2]['bi'] > bis[-4]['bi']:
zs1 = {"start_dt": bis[-2]['dt'], "end_dt": bis[-1]['dt'], "direction": "up"}
zs2 = {"start_dt": bis[-4]['dt'], "end_dt": bis[-3]['dt'], "direction": "up"}
if ka.is_bei_chi(zs1, zs2, mode="bi", adjust=0.9):
# 类趋势
if bis[-2]['bi'] > bis[-5]['bi']:
s['SBI07_最近一笔进入类趋势顶背驰段'] = True
else:
s['SBI05_最近一笔进入类盘整顶背驰段'] = True
bi_zs = find_zs(bis)
if bi_zs:
last_bi_zs = bi_zs[-1]
last_k = ka.kline_new[-1]
if last_k['close'] > last_bi_zs['ZG']:
s['SBI09_收盘价相对于最近笔中枢位置'] = 1
elif last_k['close'] < last_bi_zs['ZD']:
s['SBI09_收盘价相对于最近笔中枢位置'] = -1
else:
s['SBI09_收盘价相对于最近笔中枢位置'] = 0
if bis[-1]['fx_mark'] == 'd' and bis[-1]['bi'] > last_bi_zs['ZG']:
s["SBI10_最近笔中枢出现三买"] = True
if bis[-1]['fx_mark'] == 'g' and bis[-1]['bi'] < last_bi_zs['ZD']:
s["SBI11_最近笔中枢出现三卖"] = True
freq = ka.name
return {freq + "_" + k: v for k, v in s.items()}
def get_xd_signals_v1(ka, use_zs=False):
"""计算线段方向特征"""
s = {
"SXD01_最近一个未确认的线段标记类型": None,
"SXD02_最近一个已确认的线段标记类型": None,
"SXD03_最近一个线段内部笔标记数量": 0,
"SXD04_最近上一线段内部笔标记数量": 0,
'SXD05_最近一个线段进入类趋势顶背驰段': False,
'SXD06_最近一个线段进入类趋势底背驰段': False,
'SXD07_最近一个线段进入类盘整顶背驰段': False,
'SXD08_最近一个线段进入类盘整底背驰段': False,
"SXD09_最近向下线段不创新低": False,
"SXD10_最近向上线段不创新高": False,
"SXD11_最近两个向下线段有价格重叠区间": False,
"SXD12_最近两个向上线段有价格重叠区间": False,
}
# ------------------------------------------------------------------------------------------------------------------
if len(ka.xd_list) >= 2:
s["SXD01_最近一个未确认的线段标记类型"] = ka.xd_list[-1]['fx_mark']
s["SXD02_最近一个已确认的线段标记类型"] = ka.xd_list[-2]['fx_mark']
bi1_ = [x for x in ka.bi_list[-30:] if x['dt'] >= ka.xd_list[-1]['dt']]
s["SXD03_最近一个线段内部笔标记数量"] = len(bi1_)
bi2_ = [x for x in ka.bi_list[-60:] if ka.xd_list[-2]['dt'] <= x['dt'] <= ka.xd_list[-1]['dt']]
s["SXD04_最近上一线段内部笔标记数量"] = len(bi2_)
# ------------------------------------------------------------------------------------------------------------------
xds = ka.xd_list[-50:]
if len(xds) >= 6:
if xds[-1]['fx_mark'] == 'd' and xds[-1]['xd'] < xds[-3]['xd'] and xds[-2]['xd'] < xds[-4]['xd']:
zs1 = {"start_dt": xds[-2]['dt'], "end_dt": xds[-1]['dt'], "direction": "down"}
zs2 = {"start_dt": xds[-4]['dt'], "end_dt": xds[-3]['dt'], "direction": "down"}
if ka.is_bei_chi(zs1, zs2, mode="xd", adjust=0.9):
# 类趋势
if xds[-2]['xd'] < xds[-5]['xd']:
s['SXD06_最近一个线段进入类趋势底背驰段'] = True
else:
s['SXD08_最近一个线段进入类盘整底背驰段'] = True
if xds[-1]['fx_mark'] == 'g' and xds[-1]['xd'] > xds[-3]['xd'] and xds[-2]['xd'] > xds[-4]['xd']:
zs1 = {"start_dt": xds[-2]['dt'], "end_dt": xds[-1]['dt'], "direction": "up"}
zs2 = {"start_dt": xds[-4]['dt'], "end_dt": xds[-3]['dt'], "direction": "up"}
if ka.is_bei_chi(zs1, zs2, mode="xd", adjust=0.9):
# 类趋势
if xds[-2]['xd'] > xds[-5]['xd']:
s['SXD05_最近一个线段进入类趋势顶背驰段'] = True
else:
s['SXD07_最近一个线段进入类盘整顶背驰段'] = True
# ------------------------------------------------------------------------------------------------------------------
last_xd_inside = [x for x in ka.bi_list[-60:] if x['dt'] >= xds[-1]['dt']]
if len(xds) >= 6 and len(last_xd_inside) >= 4:
if xds[-1]['fx_mark'] == 'g':
min_bi = min(x['bi'] for x in last_xd_inside if x['fx_mark'] == 'd')
if min_bi > xds[-2]['xd']:
s["SXD09_最近向下线段不创新低"] = True
if xds[-1]['xd'] > xds[-4]['xd']:
s['SXD12_最近两个向上线段有价格重叠区间'] = True
if xds[-1]['fx_mark'] == 'd':
max_bi = max(x['bi'] for x in last_xd_inside if x['fx_mark'] == 'g')
if max_bi < xds[-2]['xd']:
s["SXD10_最近向上线段不创新高"] = True
if xds[-1]['xd'] < xds[-4]['xd']:
s["SXD11_最近两个向下线段有价格重叠区间"] = True
# ------------------------------------------------------------------------------------------------------------------
freq = ka.name
return {freq + k: v for k, v in s.items()}
class Signals:
def __init__(self, klines):
"""
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或