35 lines
1.3 KiB
Python
35 lines
1.3 KiB
Python
import numpy as np
|
|
import pandas as pd
|
|
|
|
|
|
TRAILING_HIGH_WINDOW = 60
|
|
COMPRESSION_WINDOW = 20
|
|
VOLUME_WINDOW = 20
|
|
|
|
|
|
def breakout_after_compression_score(
|
|
close: pd.DataFrame,
|
|
high: pd.DataFrame,
|
|
low: pd.DataFrame,
|
|
volume: pd.DataFrame,
|
|
) -> pd.DataFrame:
|
|
"""Score breakout setups and shift the result so it is tradable next day."""
|
|
close = close.sort_index()
|
|
high = high.reindex(index=close.index, columns=close.columns).sort_index()
|
|
low = low.reindex(index=close.index, columns=close.columns).sort_index()
|
|
volume = volume.reindex(index=close.index, columns=close.columns).sort_index()
|
|
|
|
trailing_high = close.rolling(TRAILING_HIGH_WINDOW, min_periods=TRAILING_HIGH_WINDOW).max()
|
|
proximity_to_high = close / trailing_high.replace(0, np.nan)
|
|
|
|
recent_high = high.rolling(COMPRESSION_WINDOW, min_periods=COMPRESSION_WINDOW).max()
|
|
recent_low = low.rolling(COMPRESSION_WINDOW, min_periods=COMPRESSION_WINDOW).min()
|
|
recent_mid = (recent_high + recent_low) / 2
|
|
compressed_range = -((recent_high - recent_low) / recent_mid.replace(0, np.nan))
|
|
|
|
median_volume = volume.rolling(VOLUME_WINDOW, min_periods=VOLUME_WINDOW).median()
|
|
abnormal_volume = volume / median_volume.replace(0, np.nan)
|
|
|
|
score = proximity_to_high + compressed_range + abnormal_volume
|
|
return score.shift(1)
|