Initial commit of strategies taken from swing trading dashboard
This commit is contained in:
parent
8ec1b6649a
commit
b0e84f186f
26
strategies/__init__.py
Normal file
26
strategies/__init__.py
Normal file
@ -0,0 +1,26 @@
|
|||||||
|
from .advances import advances
|
||||||
|
from .cumulative_rsi import cumulative_rsi
|
||||||
|
from .declines import declines
|
||||||
|
from .double_5s import double_5s
|
||||||
|
from .down_days_in_a_row import down_days_in_a_row
|
||||||
|
from .end_of_month import end_of_month
|
||||||
|
from .high_volume_days import high_volume_days
|
||||||
|
from .hilo_index_lows import hilo_index_lows
|
||||||
|
from .ibs_rsi import ibs_rsi
|
||||||
|
from .internal_bar_strength import internal_bar_strength
|
||||||
|
from .internal_bar_strength_band import internal_bar_strength_band
|
||||||
|
from .large_moves_down import large_moves_down
|
||||||
|
from .lower_lows import lower_lows
|
||||||
|
from .put_call_ratio_highs import put_call_ratio_highs
|
||||||
|
from .rsi_power_zones import rsi_power_zones
|
||||||
|
from .short_term_lows import short_term_lows
|
||||||
|
from .tps import tps
|
||||||
|
from .trin import trin
|
||||||
|
from .trin_thrusts import trin_thrusts
|
||||||
|
from .turnaround import turnaround
|
||||||
|
from .two_period_rsi import two_period_rsi
|
||||||
|
from .vix_above_moving_average import vix_above_moving_average
|
||||||
|
from .vix_reversal_1 import vix_reversal_1
|
||||||
|
from .vix_reversal_2 import vix_reversal_2
|
||||||
|
from .vix_reversal_3 import vix_reversal_3
|
||||||
|
from .vix_rsi import vix_rsi
|
16
strategies/advances.py
Normal file
16
strategies/advances.py
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def advances(data: DataFrame, advances: DataFrame, declines: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the Advance / Decline Ratio strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
above_200_ma = data['Close'] > data['Close'].rolling(window = 200).mean()
|
||||||
|
|
||||||
|
ratio = advances['Close'] / declines['Close']
|
||||||
|
ratio_greater_than_2 = ratio >= 2
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & ratio_greater_than_2] = 'L'
|
||||||
|
return signals
|
45
strategies/cumulative_rsi.py
Normal file
45
strategies/cumulative_rsi.py
Normal file
@ -0,0 +1,45 @@
|
|||||||
|
import pandas as pd
|
||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def calculate_moving_average(data: DataFrame, window: int = 200) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the 200-day moving average and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
return data['Close'].rolling(window = window).mean()
|
||||||
|
|
||||||
|
def calculate_rsi(data: DataFrame, period: int = 2) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the 2-period RSI and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = pd.Series(gain).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
avg_loss = pd.Series(loss).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def calculate_cumulative_rsi(rsi: Series, window: int = 2) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the cumulative RSI over a specified window period and return it as a Series.
|
||||||
|
"""
|
||||||
|
return rsi.rolling(window = window).sum()
|
||||||
|
|
||||||
|
def cumulative_rsi(data: DataFrame, rsi_period: int = 2, cumulative_period: int = 2) -> Series:
|
||||||
|
"""
|
||||||
|
Generate 'L'ong entry signals based on the Cumulative RSI strategy.
|
||||||
|
Returns a Series with 'L' for entry signals and 'N' otherwise without modifying the original DataFrame.
|
||||||
|
|
||||||
|
Entry Condition: 2-period cumulative RSI below 35 and above the 200-day moving average.
|
||||||
|
"""
|
||||||
|
ma_200 = calculate_moving_average(data)
|
||||||
|
rsi_2 = calculate_rsi(data, period = rsi_period)
|
||||||
|
cumulative_rsi_2 = calculate_cumulative_rsi(rsi_2, window = cumulative_period)
|
||||||
|
|
||||||
|
long_condition = (data['Close'] > ma_200) & (cumulative_rsi_2 < 35)
|
||||||
|
return Series(np.where(long_condition, 'L', 'N'), index = data.index)
|
16
strategies/declines.py
Normal file
16
strategies/declines.py
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def declines(data: DataFrame, declines: DataFrame, advances: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the Decline / Advance Ratio strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
above_200_ma = data['Close'] > data['Close'].rolling(window = 200).mean()
|
||||||
|
|
||||||
|
ratio = declines['Close'] / advances['Close']
|
||||||
|
ratio_greater_than_2 = ratio >= 2
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & ratio_greater_than_2] = 'L'
|
||||||
|
return signals
|
22
strategies/double_5s.py
Normal file
22
strategies/double_5s.py
Normal file
@ -0,0 +1,22 @@
|
|||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
def double_5s(data: pd.DataFrame) -> pd.Series:
|
||||||
|
"""
|
||||||
|
Generate trading signals based on the following rules:
|
||||||
|
1. SPY is above its 200-day moving average.
|
||||||
|
2. Enter a long position if SPY closes at a 5-day low.
|
||||||
|
|
||||||
|
This strategy is based on the Double 7's strategy by Larry Connors.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
ma_200 = data['Close'].rolling(window = 200).mean()
|
||||||
|
low_5_day = data['Close'].rolling(window = 5).min()
|
||||||
|
|
||||||
|
above_200_ma = data['Close'] > ma_200
|
||||||
|
at_5_day_low = data['Close'] == low_5_day
|
||||||
|
|
||||||
|
signals = pd.Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & at_5_day_low] = 'L'
|
||||||
|
|
||||||
|
return signals
|
11
strategies/down_days_in_a_row.py
Normal file
11
strategies/down_days_in_a_row.py
Normal file
@ -0,0 +1,11 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def down_days_in_a_row(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate signals for entering a long trade when the market's
|
||||||
|
close price is lower for 3 consecutive days.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
lower_close = data['Close'] < data['Close'].shift(1)
|
||||||
|
return (lower_close & lower_close.shift(1) & lower_close.shift(2)).apply(lambda x: 'L' if x else 'N')
|
8
strategies/end_of_month.py
Normal file
8
strategies/end_of_month.py
Normal file
@ -0,0 +1,8 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def end_of_month(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the End of Month (EOM) signal.
|
||||||
|
Returns a Series with 'L' for long signals if the date is close to the end of the month and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
return data['Date'].apply(lambda x: 'L' if x.day in [23, 24, 25, 26, 27] else 'N')
|
17
strategies/high_volume_days.py
Normal file
17
strategies/high_volume_days.py
Normal file
@ -0,0 +1,17 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def high_volume_days(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate swing trading signals based on the High Volume Days strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
data['200_MA'] = data['Close'].rolling(window = 200).mean()
|
||||||
|
above_200_ma = data['Close'] > data['200_MA']
|
||||||
|
|
||||||
|
highest_volume = data['Volume'] == data['Volume'].rolling(window = 5).max()
|
||||||
|
close_lower_than_open = data['Close'] < data['Open']
|
||||||
|
|
||||||
|
signal_condition = above_200_ma & highest_volume & close_lower_than_open
|
||||||
|
signals = signal_condition.apply(lambda condition: 'L' if condition else 'N')
|
||||||
|
return signals
|
16
strategies/hilo_index_lows.py
Normal file
16
strategies/hilo_index_lows.py
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def hilo_index_lows(data: DataFrame, new_highs: DataFrame, new_lows: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate signals based on the HILO Index Lows strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
above_200_ma = data['Close'] > data['Close'].rolling(window = 200).mean()
|
||||||
|
|
||||||
|
hilo_index = new_highs['Close'] - new_lows['Close']
|
||||||
|
hilo_5_day_low = hilo_index == hilo_index.rolling(window = 5).min()
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & hilo_5_day_low] = 'L'
|
||||||
|
return signals
|
35
strategies/ibs_rsi.py
Normal file
35
strategies/ibs_rsi.py
Normal file
@ -0,0 +1,35 @@
|
|||||||
|
import numpy as np
|
||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
def calculate_rsi(data: pd.DataFrame, period: int = 21) -> pd.Series:
|
||||||
|
"""
|
||||||
|
Calculate the RSI for a given period and return it as a Series.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = pd.Series(gain).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
avg_loss = pd.Series(loss).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def calculate_ibs(data: pd.DataFrame) -> pd.Series:
|
||||||
|
"""
|
||||||
|
Calculate the IBS and return it as a Series.
|
||||||
|
"""
|
||||||
|
return (data['Close'] - data['Low']) / (data['High'] - data['Low'])
|
||||||
|
|
||||||
|
def ibs_rsi(data: pd.DataFrame) -> pd.Series:
|
||||||
|
"""
|
||||||
|
Generate swing trading signals based on the IBS + RSI strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
ibs = calculate_ibs(data)
|
||||||
|
rsi_21 = calculate_rsi(data, period = 21)
|
||||||
|
|
||||||
|
conditions = (ibs < 0.25) & (rsi_21 < 45)
|
||||||
|
return pd.Series(np.where(conditions, 'L', 'N'), index = data.index)
|
9
strategies/internal_bar_strength.py
Normal file
9
strategies/internal_bar_strength.py
Normal file
@ -0,0 +1,9 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def internal_bar_strength(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the Internal Bar Strength (IBS) signal.
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
ibs_series = (data['Close'] - data['Low']) / (data['High'] - data['Low'])
|
||||||
|
return ibs_series.apply(lambda ibs: 'L' if ibs < 0.2 else 'N')
|
22
strategies/internal_bar_strength_band.py
Normal file
22
strategies/internal_bar_strength_band.py
Normal file
@ -0,0 +1,22 @@
|
|||||||
|
import pandas as pd
|
||||||
|
|
||||||
|
def internal_bar_strength_band(data: pd.DataFrame) -> pd.Series:
|
||||||
|
"""
|
||||||
|
Calculate swing trading signals based on Internal Bar Strength Band strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
ma_200 = data['Close'].rolling(window = 200).mean()
|
||||||
|
|
||||||
|
rolling_high_minus_low = (data['High'] - data['Low']).rolling(window = 25).mean()
|
||||||
|
rolling_high = data['High'].rolling(window = 10).max()
|
||||||
|
lower_band = rolling_high - 2.5 * rolling_high_minus_low
|
||||||
|
|
||||||
|
ibs = (data['Close'] - data['Low']) / (data['High'] - data['Low'])
|
||||||
|
|
||||||
|
signals = (
|
||||||
|
(data['Close'] > ma_200) &
|
||||||
|
(data['Close'] < lower_band) &
|
||||||
|
(ibs < 0.3)
|
||||||
|
).apply(lambda condition: 'L' if condition else 'N')
|
||||||
|
return signals
|
10
strategies/large_moves_down.py
Normal file
10
strategies/large_moves_down.py
Normal file
@ -0,0 +1,10 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def large_moves_down(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate swing trading signals based the Large Moves (Down) strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
percent_change = (data['Open'] - data['Close']) / data['Open']
|
||||||
|
return percent_change.apply(lambda change: 'L' if change >= 0.01 else 'N')
|
11
strategies/lower_lows.py
Normal file
11
strategies/lower_lows.py
Normal file
@ -0,0 +1,11 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def lower_lows(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate signals for entering a long trade when the market makes
|
||||||
|
a lower low for 2 consecutive days.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
lower_low = data['Low'].shift(1) > data['Low']
|
||||||
|
return (lower_low & lower_low.shift(1)).apply(lambda x: 'L' if x else 'N')
|
20
strategies/put_call_ratio_highs.py
Normal file
20
strategies/put_call_ratio_highs.py
Normal file
@ -0,0 +1,20 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def put_call_ratio_highs(data: DataFrame, put_volume: DataFrame, call_volume: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate signals based on the Put / Call Ratio Highs strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
above_200_ma = data['Close'] > data['Close'].rolling(window = 200).mean()
|
||||||
|
|
||||||
|
# The put and call volume data contains extraneous data for holidays when the market is closed.
|
||||||
|
put_volume = put_volume[put_volume['Date'].isin(data['Date'])].reset_index(drop = True)
|
||||||
|
call_volume = call_volume[call_volume['Date'].isin(data['Date'])].reset_index(drop = True)
|
||||||
|
|
||||||
|
put_call_ratio = put_volume['Close'] / call_volume['Close']
|
||||||
|
put_call_high_5_day = put_call_ratio == put_call_ratio.rolling(window = 5).max()
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & put_call_high_5_day] = 'L'
|
||||||
|
return signals
|
38
strategies/rsi_power_zones.py
Normal file
38
strategies/rsi_power_zones.py
Normal file
@ -0,0 +1,38 @@
|
|||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def calculate_moving_average(data: DataFrame, window: int = 200) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the moving average and return it as a Series.
|
||||||
|
"""
|
||||||
|
return data['Close'].rolling(window=window).mean()
|
||||||
|
|
||||||
|
def calculate_rsi(data: DataFrame, period: int = 4) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the RSI and return it as a Series.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = Series(gain).ewm(alpha=alpha, adjust=False).mean()
|
||||||
|
avg_loss = Series(loss).ewm(alpha=alpha, adjust=False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def rsi_power_zones(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate signals for the RSI PowerZones strategy.
|
||||||
|
Returns a Series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
ma_200 = calculate_moving_average(data, window = 200)
|
||||||
|
rsi_4 = calculate_rsi(data, period = 4)
|
||||||
|
|
||||||
|
above_ma_200 = data['Close'] > ma_200
|
||||||
|
rsi_below_30 = rsi_4 < 30
|
||||||
|
|
||||||
|
conditions = above_ma_200 & rsi_below_30
|
||||||
|
return Series(np.where(conditions, 'L', 'N'), index = data.index)
|
10
strategies/short_term_lows.py
Normal file
10
strategies/short_term_lows.py
Normal file
@ -0,0 +1,10 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def short_term_lows(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate signals for entering a long trade when the market makes a 10-day low.
|
||||||
|
|
||||||
|
Returns a series with 'L' for long signals and 'N' otherwise.
|
||||||
|
"""
|
||||||
|
ten_day_low = data['Low'] == data['Low'].rolling(window = 10, min_periods = 10).min()
|
||||||
|
return ten_day_low.apply(lambda x: 'L' if x else 'N')
|
40
strategies/tps.py
Normal file
40
strategies/tps.py
Normal file
@ -0,0 +1,40 @@
|
|||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def calculate_moving_average(data: DataFrame, window: int = 200) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the 200-period moving average and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
return data['Close'].rolling(window = window).mean()
|
||||||
|
|
||||||
|
def calculate_rsi(data: DataFrame, period: int = 2) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the 2-period RSI and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = Series(gain).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
avg_loss = Series(loss).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def tps(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate signals based on the Time, Price, Scale-in (TPS) strategy.
|
||||||
|
Returns a Series with 'Long' for signals and 'None' otherwise, without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
ma_200 = calculate_moving_average(data)
|
||||||
|
rsi_2 = calculate_rsi(data)
|
||||||
|
|
||||||
|
above_ma_200 = data['Close'] > ma_200
|
||||||
|
|
||||||
|
rsi_below_25 = (rsi_2 < 25)
|
||||||
|
rsi_below_25_for_two_days = rsi_below_25 & rsi_below_25.shift(1, fill_value = False)
|
||||||
|
|
||||||
|
conditions = above_ma_200 & rsi_below_25_for_two_days
|
||||||
|
return Series(np.where(conditions, 'L', 'N'), index = data.index)
|
40
strategies/trin.py
Normal file
40
strategies/trin.py
Normal file
@ -0,0 +1,40 @@
|
|||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def calculate_rsi(data: DataFrame, period: int = 2) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the RSI and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = Series(gain).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
avg_loss = Series(loss).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def trin(data: DataFrame, trin_data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the TRIN strategy with the following rules:
|
||||||
|
1. SPY is above its 200-day moving average
|
||||||
|
2. 2-period RSI is below 50
|
||||||
|
3. TRIN closes above 1
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
ma_200 = data['Close'].rolling(window = 200).mean()
|
||||||
|
|
||||||
|
rsi_2 = calculate_rsi(data, period = 2)
|
||||||
|
|
||||||
|
trin_above_1 = trin_data['Close'] > 1
|
||||||
|
|
||||||
|
above_200_ma = data['Close'] > ma_200
|
||||||
|
rsi_below_50 = rsi_2 < 50
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & rsi_below_50 & trin_above_1] = 'L'
|
||||||
|
return signals
|
15
strategies/trin_thrusts.py
Normal file
15
strategies/trin_thrusts.py
Normal file
@ -0,0 +1,15 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def trin_thrusts(data: DataFrame, trin_data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the TRIN Thrusts strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
|
||||||
|
trin_change = trin_data['Close'].pct_change()
|
||||||
|
trin_30_below = trin_change <= -0.4
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[trin_30_below] = 'L'
|
||||||
|
return signals
|
16
strategies/turnaround.py
Normal file
16
strategies/turnaround.py
Normal file
@ -0,0 +1,16 @@
|
|||||||
|
from pandas import DataFrame, Series, to_datetime
|
||||||
|
|
||||||
|
def turnaround(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the day of the week and downward close conditions.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
date_series = to_datetime(data['Date'], errors = 'coerce')
|
||||||
|
monday_or_tuesday = date_series.dt.dayofweek.isin([0, 1])
|
||||||
|
lower_past_two_days = (data['Close'] < data['Close'].shift(1)) & (data['Close'].shift(1) < data['Close'].shift(2))
|
||||||
|
|
||||||
|
signals = Series('N', index=data.index)
|
||||||
|
signals[monday_or_tuesday & lower_past_two_days] = 'L'
|
||||||
|
|
||||||
|
return signals
|
35
strategies/two_period_rsi.py
Normal file
35
strategies/two_period_rsi.py
Normal file
@ -0,0 +1,35 @@
|
|||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def calculate_moving_average(data: DataFrame, window: int = 200) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the 200-period moving average and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
return data['Close'].rolling(window = window).mean()
|
||||||
|
|
||||||
|
def calculate_rsi(data: DataFrame, period: int = 2) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the 2-period RSI and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = Series(gain).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
avg_loss = Series(loss).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def two_period_rsi(data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate signals based on the 200-period MA and 2-period RSI.
|
||||||
|
Returns a Series with 'Long' for signals and 'None' otherwise, without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
ma_200 = calculate_moving_average(data)
|
||||||
|
rsi_2 = calculate_rsi(data)
|
||||||
|
|
||||||
|
conditions = (data['Close'] > ma_200) & (rsi_2 < 15)
|
||||||
|
return Series(np.where(conditions, 'L', 'N'), index = data.index)
|
17
strategies/vix_above_moving_average.py
Normal file
17
strategies/vix_above_moving_average.py
Normal file
@ -0,0 +1,17 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def vix_above_moving_average(data: DataFrame, vix_data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the VIX Above Moving Average strategy.
|
||||||
|
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
ma_200 = data['Close'].rolling(window = 200).mean()
|
||||||
|
above_200_ma = data['Close'] > ma_200
|
||||||
|
|
||||||
|
vix_ma_10 = vix_data['Close'].rolling(window = 10).mean()
|
||||||
|
vix_close_10_percent_above_ma = vix_data['Close'] >= (1.1 * vix_ma_10)
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & vix_close_10_percent_above_ma] = 'L'
|
||||||
|
return signals
|
17
strategies/vix_reversal_1.py
Normal file
17
strategies/vix_reversal_1.py
Normal file
@ -0,0 +1,17 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def vix_reversal_1(data: DataFrame, vix_data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the VIX 15-day high strategy.
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
ma_200 = data['Close'].rolling(window = 200).mean()
|
||||||
|
above_200_ma = data['Close'] > ma_200
|
||||||
|
|
||||||
|
vix_15_day_high = vix_data['High'] == vix_data['High'].rolling(window = 15).max()
|
||||||
|
|
||||||
|
vix_close_below_open = vix_data['Close'] < vix_data['Open']
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & vix_15_day_high & vix_close_below_open] = 'L'
|
||||||
|
return signals
|
32
strategies/vix_reversal_2.py
Normal file
32
strategies/vix_reversal_2.py
Normal file
@ -0,0 +1,32 @@
|
|||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def calculate_rsi(data: DataFrame, period: int = 5) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the RSI and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = Series(gain).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
avg_loss = Series(loss).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def vix_reversal_2(data: DataFrame, vix_data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate swing trading signals based on the Connors VIX Reversal 2 strategy.
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
vix_rsi_5 = calculate_rsi(vix_data, period = 5)
|
||||||
|
|
||||||
|
vix_rsi_above_70 = vix_rsi_5.shift(1) > 70
|
||||||
|
vix_close_below_prev_close = vix_data['Close'] < vix_data['Close'].shift(1)
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[vix_rsi_above_70 & vix_close_below_prev_close] = 'L'
|
||||||
|
return signals
|
15
strategies/vix_reversal_3.py
Normal file
15
strategies/vix_reversal_3.py
Normal file
@ -0,0 +1,15 @@
|
|||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def vix_reversal_3(data: DataFrame, vix_data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the Connors VIX Reversal 3 strategy.
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
vix_ma_50 = vix_data['Close'].rolling(window = 50).mean()
|
||||||
|
|
||||||
|
vix_low_above_50_ma = vix_data['Low'] > vix_ma_50
|
||||||
|
vix_close_10_percent_above_ma = vix_data['Close'] >= (1.1 * vix_ma_50)
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[vix_low_above_50_ma & vix_close_10_percent_above_ma] = 'L'
|
||||||
|
return signals
|
38
strategies/vix_rsi.py
Normal file
38
strategies/vix_rsi.py
Normal file
@ -0,0 +1,38 @@
|
|||||||
|
import numpy as np
|
||||||
|
|
||||||
|
from pandas import DataFrame, Series
|
||||||
|
|
||||||
|
def calculate_rsi(data: DataFrame, period: int = 2) -> Series:
|
||||||
|
"""
|
||||||
|
Calculate the RSI and return it as a Series without modifying the original DataFrame.
|
||||||
|
"""
|
||||||
|
delta = data['Close'].diff()
|
||||||
|
gain = np.where(delta > 0, delta, 0)
|
||||||
|
loss = np.where(delta < 0, -delta, 0)
|
||||||
|
|
||||||
|
alpha = 1 / period
|
||||||
|
avg_gain = Series(gain).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
avg_loss = Series(loss).ewm(alpha = alpha, adjust = False).mean()
|
||||||
|
|
||||||
|
rs = avg_gain / avg_loss
|
||||||
|
return 100 - (100 / (1 + rs))
|
||||||
|
|
||||||
|
def vix_rsi(data: DataFrame, vix_data: DataFrame) -> Series:
|
||||||
|
"""
|
||||||
|
Generate long signals based on the VIX RSI strategy.
|
||||||
|
Returns a Series with 'L' for long signals and 'N' for no signal.
|
||||||
|
"""
|
||||||
|
ma_200 = data['Close'].rolling(window = 200).mean()
|
||||||
|
rsi_2 = calculate_rsi(data, period = 2)
|
||||||
|
|
||||||
|
vix_rsi_2 = calculate_rsi(vix_data, period = 2)
|
||||||
|
|
||||||
|
above_200_ma = data['Close'] > ma_200
|
||||||
|
vix_rsi_above_90 = vix_rsi_2 > 90
|
||||||
|
vix_open_greater_than_prev_close = vix_data['Open'] > vix_data['Close'].shift(1)
|
||||||
|
rsi_below_30 = rsi_2 < 30
|
||||||
|
|
||||||
|
signals = Series('N', index = data.index)
|
||||||
|
signals[above_200_ma & vix_rsi_above_90 & vix_open_greater_than_prev_close & rsi_below_30] = 'L'
|
||||||
|
|
||||||
|
return signals
|
Loading…
Reference in New Issue
Block a user