class ArrayManager(object):
"""
For:
vnpy 2.6.0 新增部分函数
"""
def sar(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.SAR(self.high, self.low, acceleration=0, maximum=0)
if array:
return result
return result[-1]
def t3(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
result = talib.T3(self.high, n, vfactor=0)
if array:
return result
return result[-1]
def dema(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
result = talib.DEMA(self.close, n)
if array:
return result
return result[-1]
def tema(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
result = talib.TEMA(self.close, n)
if array:
return result
return result[-1]
def midpoint(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
result = talib.MIDPOINT(self.close, n)
if array:
return result
return result[-1]
def midprice(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
result = talib.MIDPRICE(self.high, self.low, n)
if array:
return result
return result[-1]
def trima(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
result = talib.TRIMA(self.close, n)
if array:
return result
return result[-1]
def ht_dcperiod(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.HT_DCPERIOD(self.close)
if array:
return result
return result[-1]
def ht_dcphase(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.HT_DCPHASE(self.close)
if array:
return result
return result[-1]
def ht_trendmode(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.HT_TRENDMODE(self.close)
if array:
return result
return result[-1]
def avgprice(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.$GPRICE(self.open,self.high,self.low,self.close)
if array:
return result
return result[-1]
def medprice(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.MEDPRICE(self.high, self.low)
if array:
return result
return result[-1]
def typprice(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.TYPPRICE(self.high, self.low, self.close)
if array:
return result
return result[-1]
def wclprice(self, array: bool = False) -> Union[float, np.ndarray]:
result = talib.WCLPRICE(self.high, self.low, self.close)
if array:
return result
return result[-1]
def stoch(self,fastk_period:int,slowk_period:int,slowk_matype:int,slowd_period:int,slowd_matype:int,array: bool = False) -> Union[float, np.ndarray]:
STOCH_slowk, STOCH_slowd = talib.STOCH(self.high, self.low, self.close,
fastk_period=5, slowk_period=3, slowk_matype=0,
slowd_period=3, slowd_matype=0)
if array:
return STOCH_slowk,STOCH_slowd
return STOCH_slowk[-1],STOCH_slowd[-1]
def stochf(self, fastk_period:int ,fastd_period:int,fastd_matype:int,array: bool = False) -> Union[float, np.ndarray]:
STOCHF_fastk, STOCHF_fastd = talib.STOCHF(self.high, self.low, self.close,
fastk_period=5, fastd_period=3, fastd_matype=0)
if array:
return STOCHF_fastk, STOCHF_fastd
return STOCHF_fastk[-1], STOCHF_fastd[-1]
def beta(self, n: int, array: bool = False) -> Union[float, np.ndarray]:
result = talib.BETA(self.high, self.low, n)
if array:
return result
return result[-1]
def correl(self, n: int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.CORREL(self.high,self.low, n)
if array:
return result
return result[-1]
def linearreg(self, n: int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.LINEARREG(self.close, n)
if array:
return result
return result[-1]
def linearreg_angle(self, n: int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.LINEARREG_ANGLE(self.close, n)
if array:
return result
return result[-1]
def linearreg_intercept(self, n: int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.LINEARREG_INTERCEPT(self.close, n)
if array:
return result
return result[-1]
def linearreg_slope(self, n: int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.LINEARREG_SLOPE(self.close, n)
if array:
return result
return result[-1]
def tsf(self, n: int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.TSF(self.close, n)
if array:
return result
return result[-1]
def stddev(self, timeperiod: int,nbdev:int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.STDDEV(self.close, timeperiod, nbdev)
if array:
return result
return result[-1]
def var(self, timeperiod: int,nbdev:int ,array: bool = False) -> Union[float, np.ndarray]:
result = talib.VAR(self.close, timeperiod, nbdev)
if array:
return result
return result[-1]
def ht_phasor(self, array: bool = False) -> Union[float, np.ndarray]:
HT_PHASOR_inphase,HT_PHASOR_quadrature = talib.HT_PHASOR(self.close)
if array:
return HT_PHASOR_inphase,HT_PHASOR_quadrature
return HT_PHASOR_inphase[-1],HT_PHASOR_quadrature[-1]
def ht_sine(self, array: bool = False) -> Union[float, np.ndarray]:
HT_SINE_sine,HT_SINE_leadsine = talib.HT_SINE(self.close)
if array:
return HT_SINE_sine,HT_SINE_leadsine
return HT_SINE_sine[-1],HT_SINE_leadsine[-1]
def sarext(self,
startvalue: int,
offsetonreverse: int,
accelerationinitlong: int,
accelerationlong: int,
accelerationmaxlong: int,
accelerationinitshort: int,
accelerationshort: int,
accelerationmaxshort: int,
array: bool = False) -> Union[float, np.ndarray]:
result = talib.SAREXT(self.high, self.low, startvalue, offsetonreverse,
accelerationinitlong, accelerationlong, accelerationmaxlong,
accelerationinitshort, accelerationshort, accelerationmaxshort)
if array:
return result
return result[-1]
def bbands(
self,
array: bool = False
) -> Union[
Tuple[np.ndarray, np.ndarray, np.ndarray],
Tuple[float, float, float]
]:
BBANDS_upper, BBANDS_middle, BBANDS_lower = talib.BBANDS(
self.close, matype=MA_Type.T3
)
if array:
return BBANDS_upper, BBANDS_middle, BBANDS_lower
return BBANDS_upper[-1], BBANDS_middle[-1], BBANDS_lower[-1]
def macdext(
self,
fast_period: int,
fast_matype: int,
slow_period: int,
slow_matype: int,
signal_period: int,
signal_matype: int,
array: bool = False
) -> Union[
Tuple[np.ndarray, np.ndarray, np.ndarray],
Tuple[float, float, float]
]:
macd, signal, hist = talib.MACDEXT(
self.close, fast_period, fast_matype, slow_period, slow_matype, signal_period, signal_matype
)
if array:
return macd, signal, hist
return macd[-1], signal[-1], hist[-1]
def macdfix(
self,
signal_period: int,
array: bool = False
) -> Union[
Tuple[np.ndarray, np.ndarray, np.ndarray],
Tuple[float, float, float]
]:
macd, signal, hist = talib.MACDFIX(
self.close, signal_period
)
if array:
return macd, signal, hist
return macd[-1], signal[-1], hist[-1]
