forked from RoveAllOverTheWorld512/hyb_ta
-
Notifications
You must be signed in to change notification settings - Fork 1
/
hhvbars.py
76 lines (61 loc) · 1.89 KB
/
hhvbars.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
# -*- coding: utf-8 -*-
"""
Created on 2019-10-21 13:33:16
author: huangyunbin
email: huangyunbin@sina.com
QQ: 592440193
"""
import pandas as pd
import numpy as np
from math import isnan
def idxmax(ds,i,l):
'''
返回pandas.Series指定位置、指定区间内的最大值及索引
注意:ds索引号为顺序号,l>0
'''
if isnan(i):
return None
if (l <= 0) | ((i - l + 1) < 0) | (i > len(ds)):
return None
s = ds.iloc[(i - l + 1): i + 1]
return s.loc[[s.idxmax()]]
def idxmin(ds, i, l):
'''
返回pandas.Series指定位置、指定区间内的最小值及索引
注意:ds索引号为顺序号,l>0
'''
if isnan(i):
return None
if (l <= 0) | ((i - l + 1) < 0) | (i > len(ds)):
return None
# print(f'i={i}')
s = ds.iloc[(i - l + 1): i + 1]
return s.loc[[s.idxmin()]]
if isinstance(ds, pd.Series):
name = ds.name
if not isinstance(ds.index, pd.RangeIndex):
ds1 = ds.reset_index()
ds2 = ds1[name]
n = 34
ds3 = pd.concat([idxmax(ds2, i, n) for i in range(len(ds2))])
ds3 = ds3.reset_index()
ds3.index += (n - 1) # 由于前面返回
ds1['max'] = ds3['close']
ds1['idxmax'] = ds3['index']
m = 34
ds4 = pd.concat([(idxmin(ds2, int(ds1.iloc[i]['idxmax']), m)
if not isnan(ds1.iloc[i]['idxmax']) else None)
for i in range(len(ds2))])
ds4 = ds4.reset_index()
ds4.index += (len(ds2)-len(ds4)) # 由于前面返回
ds1['min'] = ds4['close']
ds1['idxmin'] = ds4['index']
ds1 = ds1.assign(decreasing=ds1['close']/ds1['max']-1)
ds1 = ds1.assign(increasing=ds1['max']/ds1['min']-1)
in_threshold = 0.25
de_threshold = -0.15
ds1 = ds1.assign(double_bott=((ds1['decreasing'] < de_threshold)
& (ds1['increasing'] > in_threshold)))
ds1['double_bott'] = ds1['double_bott'].astype(int)
ds1['double_bott'] = ds1['double_bott'].replace(0, np.NaN)
ds1.to_csv('tmp.csv')