Module technical
Expand source code
import numpy as np
import pandas as pd
from tqdm.auto import tqdm
from utils import ultimate_cycle
# OSCILLATORS
def macd(prices: pd.Series, long: int, short: int, strategy=False, getgains=False, winning=False, commissions=0.005) -> pd.Series:
'''
Return the MACD
:param pd.Series prices: Prices of the stock
:param int long: Long moving average length
:param int short: Short moving average length
:param bool strategy: If strategy should be returned
:param bool getgains: If gains should be returned
:param bool winning: If policy gain - no strategy gain should be returned
:param float commissions: Percentage commissions per transaction
'''
if prices.index.duplicated().any():
raise ValueError("There are some duplicate indexes.")
macdvalues = prices.rolling(short).mean() - prices.rolling(long).mean()
if winning:
positive = macdvalues > 0
policy = positive.shift(1) != positive
if positive.iloc[0]:
policy.iloc[0] = 1
gain = gains(prices=prices, policy=policy, commissions=commissions)
diff = (prices.iloc[-1]/prices.iloc[0]) - 1
return gain.sum() - diff * 100
if strategy:
positive = macdvalues > 0
return positive.shift(1) != positive
if getgains:
positive = macdvalues > 0
policy = positive.shift(1) != positive
if positive.iloc[0]:
policy.iloc[0] = 1
return gains(prices=prices, policy=policy, commissions=commissions)
return macdvalues
def ultimate(prices: pd.Series, low: pd.Series, high: pd.Series, buylevel=30, selllevel=70, days=7, strategy=False, getgains=False, winning=False, commissions=0.005, mingain=0, accelerate=True, firstopportunity=False, stoploss=0) -> pd.Series:
'''
Return the Ultimate oscillator
:param pd.Series prices: Prices of the stock
:param pd.Series low: Long moving average length
:param pd.Series high: Short moving average length
:param int days: Days for moving sum
:param bool strategy: If strategy should be returned
:param bool getgains: If gains should be returned
:param bool winning: If policy gain - no strategy gain should be returned
:param float commissions: Percentage commissions per transaction
:param bool accelerate: If uses cython
:param float mingain: Minimum gain to sell
:param bool firstopportunity: If sell first time you have mingain
:param float stoploss: Maximum percentage loss
'''
if prices.index.duplicated().any():
raise ValueError("There are some duplicate indexes.")
bp = prices - np.minimum(prices.shift(1), low)
tr = np.maximum(high, prices.shift(1)) - np.minimum(prices.shift(1), low)
avg1 = bp.rolling(days).sum()/tr.rolling(days).sum()
avg2 = bp.rolling(2*days).sum()/tr.rolling(2*days).sum()
avg3 = bp.rolling(3*days).sum()/tr.rolling(3*days).sum()
ult = 100 * (4*avg1 + 2*avg2 + avg3)/7
if winning or strategy or getgains:
buy = ult < buylevel
sell = ult > selllevel
policy = getpolicy(buy=buy, sell=sell, prices=prices, mingain=mingain, stoploss=stoploss, accelerate=accelerate, firstopportunity=firstopportunity)
else:
return ult
if winning:
gain = gains(prices=prices, policy=policy, commissions=commissions)
diff = (prices.iloc[-1]/prices.iloc[0]) - 1
return gain.sum() - diff * 100
if strategy:
return policy
if getgains:
return gains(prices=prices, policy=policy, commissions=commissions)
def bollinger_bands(prices: pd.Series, k=1, period=1000, strategy=False, getgains=False, winning=False, commissions=0.005, accelerate=True, mingain=0, firstopportunity=False, stoploss=0) -> (pd.Series, pd.Series):
'''
Return the Bollinger bands
:param pd.Series prices: Prices of the stock
:param int k: How many standard deviations out
:param int period: Period for moving average
:param bool strategy: If strategy should be returned
:param bool getgains: If gains should be returned
:param bool winning: If policy gain - no strategy gain should be returned
:param float commissions: Percentage commissions per transaction
:param bool accelerate: If uses cython
:param float mingain: Minimum gain to sell
:param bool firstopportunity: If sell first time you have mingain
:param float stoploss: Maximum percentage loss
'''
std = prices.rolling(period).std()
mean = prices.rolling(period).mean()
upperband = mean + std*k
lowerband = mean - std*k
if strategy or getgains or winning:
sell = prices > upperband
buy = prices < lowerband
policy = getpolicy(buy=buy, sell=sell, prices=prices, mingain=mingain, stoploss=stoploss, accelerate=accelerate, firstopportunity=firstopportunity)
if winning:
gain = gains(prices=prices, policy=policy, commissions=commissions)
diff = (prices.iloc[-1]/prices.iloc[0]) - 1
return gain.sum() - diff * 100
if strategy:
return policy
if getgains:
return gains(prices=prices, policy=policy, commissions=commissions)
return lowerband, upperband
# UTILS
def gains(prices: pd.Series, policy: pd.Series, budget=100, commissions=0.005) -> pd.Series:
'''
Return the gains
:param pd.Series prices: Prices of the stock
:param pd.Series policy: True when buy or sell
:param float budget: My budget
:param float commissions: Percentage commissions per transaction
'''
prices = prices.loc[policy.index]
buy = prices[policy].iloc[::2]
sell = prices[policy].iloc[1::2]
buy = buy.iloc[:sell.size].values
gains = (sell/buy) - 1
return (gains - commissions*2)*budget
def getpolicy(buy: pd.Series, sell: pd.Series, prices: pd.Series, mingain=0, stoploss=0, accelerate=True, firstopportunity=False) -> pd.Series:
"""
Return the policy given all the moments sell or buy is True
:param pd.Series buy: When the buy pricinple is respected
:param pd.Series sell: When the sell pricinple is respected
:param float mingain: Minimum gain to sell
:param float stoploss: Maximum percentage loss
:param bool accelerate: If use cython
:param bool firstopportunity: If sell first time you have mingain, MUST USE ACCELERATE
"""
if firstopportunity and not accelerate:
print("Chaning accelerate to True to use firstopportunity.")
accelerate = True
buys = buy.shift(1) != buy
sells = sell.shift(1) != sell
policy = pd.Series(np.zeros(buy.size), index=buy.index)
if accelerate:
buys.reset_index(drop=True, inplace=True)
sells.reset_index(drop=True, inplace=True)
index = buys[buys | sells].index.to_numpy()
if mingain == 0 and stoploss == 0:
policy_ = ultimate_cycle.ultimate_cycle(policy.to_numpy(), buys.to_numpy(), sells.to_numpy(), index)
elif not firstopportunity and stoploss == 0:
policy_ = ultimate_cycle.cycle_checkgain(policy.to_numpy(), buys.to_numpy(), sells.to_numpy(), index, prices.to_numpy(), mingain)
else:
policy_ = ultimate_cycle.cycle_absolutegain(policy.to_numpy(dtype=bool), buys.to_numpy(dtype=bool), buys[buys].index.to_numpy(dtype=np.int32), prices.to_numpy(dtype=np.float32), mingain, stoploss)
policy = pd.Series(policy_, index=policy.index)
else:
token = 1
buy_price = 0
for idx in tqdm(buys[buys | sells].index):
if token and buys.loc[idx]:
policy.loc[idx] = 1
token = 0
buy_price = prices.loc[idx]
elif not token and sells.loc[idx] and mingain*(prices.loc[idx]/buy_price) >= mingain*(1 + mingain):
policy.loc[idx] = 1
token = 1
return policy == 1
Functions
def bollinger_bands(prices: pandas.core.series.Series, k=1, period=1000, strategy=False, getgains=False, winning=False, commissions=0.005, accelerate=True, mingain=0, firstopportunity=False, stoploss=0) ‑> (
, ) -
Return the Bollinger bands
:param pd.Series prices: Prices of the stock :param int k: How many standard deviations out :param int period: Period for moving average :param bool strategy: If strategy should be returned :param bool getgains: If gains should be returned :param bool winning: If policy gain - no strategy gain should be returned :param float commissions: Percentage commissions per transaction :param bool accelerate: If uses cython :param float mingain: Minimum gain to sell :param bool firstopportunity: If sell first time you have mingain :param float stoploss: Maximum percentage loss
Expand source code
def bollinger_bands(prices: pd.Series, k=1, period=1000, strategy=False, getgains=False, winning=False, commissions=0.005, accelerate=True, mingain=0, firstopportunity=False, stoploss=0) -> (pd.Series, pd.Series): ''' Return the Bollinger bands :param pd.Series prices: Prices of the stock :param int k: How many standard deviations out :param int period: Period for moving average :param bool strategy: If strategy should be returned :param bool getgains: If gains should be returned :param bool winning: If policy gain - no strategy gain should be returned :param float commissions: Percentage commissions per transaction :param bool accelerate: If uses cython :param float mingain: Minimum gain to sell :param bool firstopportunity: If sell first time you have mingain :param float stoploss: Maximum percentage loss ''' std = prices.rolling(period).std() mean = prices.rolling(period).mean() upperband = mean + std*k lowerband = mean - std*k if strategy or getgains or winning: sell = prices > upperband buy = prices < lowerband policy = getpolicy(buy=buy, sell=sell, prices=prices, mingain=mingain, stoploss=stoploss, accelerate=accelerate, firstopportunity=firstopportunity) if winning: gain = gains(prices=prices, policy=policy, commissions=commissions) diff = (prices.iloc[-1]/prices.iloc[0]) - 1 return gain.sum() - diff * 100 if strategy: return policy if getgains: return gains(prices=prices, policy=policy, commissions=commissions) return lowerband, upperband
def gains(prices: pandas.core.series.Series, policy: pandas.core.series.Series, budget=100, commissions=0.005) ‑> pandas.core.series.Series
-
Return the gains
:param pd.Series prices: Prices of the stock :param pd.Series policy: True when buy or sell :param float budget: My budget :param float commissions: Percentage commissions per transaction
Expand source code
def gains(prices: pd.Series, policy: pd.Series, budget=100, commissions=0.005) -> pd.Series: ''' Return the gains :param pd.Series prices: Prices of the stock :param pd.Series policy: True when buy or sell :param float budget: My budget :param float commissions: Percentage commissions per transaction ''' prices = prices.loc[policy.index] buy = prices[policy].iloc[::2] sell = prices[policy].iloc[1::2] buy = buy.iloc[:sell.size].values gains = (sell/buy) - 1 return (gains - commissions*2)*budget
def getpolicy(buy: pandas.core.series.Series, sell: pandas.core.series.Series, prices: pandas.core.series.Series, mingain=0, stoploss=0, accelerate=True, firstopportunity=False) ‑> pandas.core.series.Series
-
Return the policy given all the moments sell or buy is True
:param pd.Series buy: When the buy pricinple is respected :param pd.Series sell: When the sell pricinple is respected :param float mingain: Minimum gain to sell :param float stoploss: Maximum percentage loss :param bool accelerate: If use cython :param bool firstopportunity: If sell first time you have mingain, MUST USE ACCELERATE
Expand source code
def getpolicy(buy: pd.Series, sell: pd.Series, prices: pd.Series, mingain=0, stoploss=0, accelerate=True, firstopportunity=False) -> pd.Series: """ Return the policy given all the moments sell or buy is True :param pd.Series buy: When the buy pricinple is respected :param pd.Series sell: When the sell pricinple is respected :param float mingain: Minimum gain to sell :param float stoploss: Maximum percentage loss :param bool accelerate: If use cython :param bool firstopportunity: If sell first time you have mingain, MUST USE ACCELERATE """ if firstopportunity and not accelerate: print("Chaning accelerate to True to use firstopportunity.") accelerate = True buys = buy.shift(1) != buy sells = sell.shift(1) != sell policy = pd.Series(np.zeros(buy.size), index=buy.index) if accelerate: buys.reset_index(drop=True, inplace=True) sells.reset_index(drop=True, inplace=True) index = buys[buys | sells].index.to_numpy() if mingain == 0 and stoploss == 0: policy_ = ultimate_cycle.ultimate_cycle(policy.to_numpy(), buys.to_numpy(), sells.to_numpy(), index) elif not firstopportunity and stoploss == 0: policy_ = ultimate_cycle.cycle_checkgain(policy.to_numpy(), buys.to_numpy(), sells.to_numpy(), index, prices.to_numpy(), mingain) else: policy_ = ultimate_cycle.cycle_absolutegain(policy.to_numpy(dtype=bool), buys.to_numpy(dtype=bool), buys[buys].index.to_numpy(dtype=np.int32), prices.to_numpy(dtype=np.float32), mingain, stoploss) policy = pd.Series(policy_, index=policy.index) else: token = 1 buy_price = 0 for idx in tqdm(buys[buys | sells].index): if token and buys.loc[idx]: policy.loc[idx] = 1 token = 0 buy_price = prices.loc[idx] elif not token and sells.loc[idx] and mingain*(prices.loc[idx]/buy_price) >= mingain*(1 + mingain): policy.loc[idx] = 1 token = 1 return policy == 1
def macd(prices: pandas.core.series.Series, long: int, short: int, strategy=False, getgains=False, winning=False, commissions=0.005) ‑> pandas.core.series.Series
-
Return the MACD
:param pd.Series prices: Prices of the stock :param int long: Long moving average length :param int short: Short moving average length :param bool strategy: If strategy should be returned :param bool getgains: If gains should be returned :param bool winning: If policy gain - no strategy gain should be returned :param float commissions: Percentage commissions per transaction
Expand source code
def macd(prices: pd.Series, long: int, short: int, strategy=False, getgains=False, winning=False, commissions=0.005) -> pd.Series: ''' Return the MACD :param pd.Series prices: Prices of the stock :param int long: Long moving average length :param int short: Short moving average length :param bool strategy: If strategy should be returned :param bool getgains: If gains should be returned :param bool winning: If policy gain - no strategy gain should be returned :param float commissions: Percentage commissions per transaction ''' if prices.index.duplicated().any(): raise ValueError("There are some duplicate indexes.") macdvalues = prices.rolling(short).mean() - prices.rolling(long).mean() if winning: positive = macdvalues > 0 policy = positive.shift(1) != positive if positive.iloc[0]: policy.iloc[0] = 1 gain = gains(prices=prices, policy=policy, commissions=commissions) diff = (prices.iloc[-1]/prices.iloc[0]) - 1 return gain.sum() - diff * 100 if strategy: positive = macdvalues > 0 return positive.shift(1) != positive if getgains: positive = macdvalues > 0 policy = positive.shift(1) != positive if positive.iloc[0]: policy.iloc[0] = 1 return gains(prices=prices, policy=policy, commissions=commissions) return macdvalues
def ultimate(prices: pandas.core.series.Series, low: pandas.core.series.Series, high: pandas.core.series.Series, buylevel=30, selllevel=70, days=7, strategy=False, getgains=False, winning=False, commissions=0.005, mingain=0, accelerate=True, firstopportunity=False, stoploss=0) ‑> pandas.core.series.Series
-
Return the Ultimate oscillator
:param pd.Series prices: Prices of the stock :param pd.Series low: Long moving average length :param pd.Series high: Short moving average length :param int days: Days for moving sum :param bool strategy: If strategy should be returned :param bool getgains: If gains should be returned :param bool winning: If policy gain - no strategy gain should be returned :param float commissions: Percentage commissions per transaction :param bool accelerate: If uses cython :param float mingain: Minimum gain to sell :param bool firstopportunity: If sell first time you have mingain :param float stoploss: Maximum percentage loss
Expand source code
def ultimate(prices: pd.Series, low: pd.Series, high: pd.Series, buylevel=30, selllevel=70, days=7, strategy=False, getgains=False, winning=False, commissions=0.005, mingain=0, accelerate=True, firstopportunity=False, stoploss=0) -> pd.Series: ''' Return the Ultimate oscillator :param pd.Series prices: Prices of the stock :param pd.Series low: Long moving average length :param pd.Series high: Short moving average length :param int days: Days for moving sum :param bool strategy: If strategy should be returned :param bool getgains: If gains should be returned :param bool winning: If policy gain - no strategy gain should be returned :param float commissions: Percentage commissions per transaction :param bool accelerate: If uses cython :param float mingain: Minimum gain to sell :param bool firstopportunity: If sell first time you have mingain :param float stoploss: Maximum percentage loss ''' if prices.index.duplicated().any(): raise ValueError("There are some duplicate indexes.") bp = prices - np.minimum(prices.shift(1), low) tr = np.maximum(high, prices.shift(1)) - np.minimum(prices.shift(1), low) avg1 = bp.rolling(days).sum()/tr.rolling(days).sum() avg2 = bp.rolling(2*days).sum()/tr.rolling(2*days).sum() avg3 = bp.rolling(3*days).sum()/tr.rolling(3*days).sum() ult = 100 * (4*avg1 + 2*avg2 + avg3)/7 if winning or strategy or getgains: buy = ult < buylevel sell = ult > selllevel policy = getpolicy(buy=buy, sell=sell, prices=prices, mingain=mingain, stoploss=stoploss, accelerate=accelerate, firstopportunity=firstopportunity) else: return ult if winning: gain = gains(prices=prices, policy=policy, commissions=commissions) diff = (prices.iloc[-1]/prices.iloc[0]) - 1 return gain.sum() - diff * 100 if strategy: return policy if getgains: return gains(prices=prices, policy=policy, commissions=commissions)