feat: Add signal dates to ATR EMA screeners for backtesting
This commit is contained in:
parent
a8dd33c3d9
commit
2f784e5d65
@ -24,7 +24,7 @@ def check_atr_ema_bullish_signal(df: pd.DataFrame) -> bool:
|
|||||||
print(f"Lower Band: ${last_bands['lower_band']:.2f}")
|
print(f"Lower Band: ${last_bands['lower_band']:.2f}")
|
||||||
print(f"Bullish Signal: {'Yes' if last_bands['signal'] else 'No'}")
|
print(f"Bullish Signal: {'Yes' if last_bands['signal'] else 'No'}")
|
||||||
|
|
||||||
def check_atr_ema_buy_condition(df: pd.DataFrame) -> bool:
|
def check_atr_ema_buy_condition(df: pd.DataFrame) -> tuple:
|
||||||
"""Check if price is below EMA and moving up through lower ATR band"""
|
"""Check if price is below EMA and moving up through lower ATR band"""
|
||||||
# Get latest values from DataFrame
|
# Get latest values from DataFrame
|
||||||
last_price = df.iloc[-1]
|
last_price = df.iloc[-1]
|
||||||
@ -35,12 +35,14 @@ def check_atr_ema_buy_condition(df: pd.DataFrame) -> bool:
|
|||||||
ema = results['ema'].iloc[-1]
|
ema = results['ema'].iloc[-1]
|
||||||
lower_band = results['lower_band'].iloc[-1]
|
lower_band = results['lower_band'].iloc[-1]
|
||||||
|
|
||||||
return (
|
signal = (
|
||||||
last_price['close'] < ema and
|
last_price['close'] < ema and
|
||||||
previous_price['close'] <= lower_band and
|
previous_price['close'] <= lower_band and
|
||||||
last_price['close'] > previous_price['close']
|
last_price['close'] > previous_price['close']
|
||||||
)
|
)
|
||||||
|
|
||||||
|
return signal, last_price['date'] if signal else None, results.iloc[-1]
|
||||||
|
|
||||||
def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None:
|
def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None:
|
||||||
print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}")
|
print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}")
|
||||||
|
|
||||||
@ -108,8 +110,9 @@ def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, por
|
|||||||
results = indicator.calculate(df)
|
results = indicator.calculate(df)
|
||||||
|
|
||||||
# Check for signals
|
# Check for signals
|
||||||
if results['bullish_signal'].iloc[-1]:
|
signal, signal_date, indicator_values = check_atr_ema_buy_condition(df)
|
||||||
target_price = results['upper_band'].iloc[-1]
|
if signal:
|
||||||
|
target_price = indicator_values['upper_band']
|
||||||
|
|
||||||
if calculator:
|
if calculator:
|
||||||
position = calculator.calculate_position_size(current_price, target_price)
|
position = calculator.calculate_position_size(current_price, target_price)
|
||||||
@ -118,6 +121,7 @@ def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, por
|
|||||||
'ticker': ticker,
|
'ticker': ticker,
|
||||||
'entry': current_price,
|
'entry': current_price,
|
||||||
'target': target_price,
|
'target': target_price,
|
||||||
|
'signal_date': signal_date,
|
||||||
'volume': current_volume,
|
'volume': current_volume,
|
||||||
'last_update': datetime.fromtimestamp(last_update/1000000000),
|
'last_update': datetime.fromtimestamp(last_update/1000000000),
|
||||||
'shares': position['shares'],
|
'shares': position['shares'],
|
||||||
@ -129,7 +133,7 @@ def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, por
|
|||||||
}
|
}
|
||||||
bullish_signals.append(signal_data)
|
bullish_signals.append(signal_data)
|
||||||
dollar_risk = signal_data['risk'] * -1
|
dollar_risk = signal_data['risk'] * -1
|
||||||
print(f"\n🟢 {ticker} @ ${current_price:.2f}")
|
print(f"\n🟢 {ticker} @ ${current_price:.2f} on {signal_date.strftime('%Y-%m-%d %H:%M')}")
|
||||||
print(f" Size: {signal_data['shares']} shares (${signal_data['position_size']:.2f})")
|
print(f" Size: {signal_data['shares']} shares (${signal_data['position_size']:.2f})")
|
||||||
print(f" Stop: ${signal_data['stop_loss']:.2f} (7%) | Target: ${target_price:.2f}")
|
print(f" Stop: ${signal_data['stop_loss']:.2f} (7%) | Target: ${target_price:.2f}")
|
||||||
print(f" Risk/Reward: 1:{signal_data['r_r']:.1f} | Risk: ${dollar_risk:.2f}")
|
print(f" Risk/Reward: 1:{signal_data['r_r']:.1f} | Risk: ${dollar_risk:.2f}")
|
||||||
|
|||||||
@ -7,7 +7,7 @@ from utils.data_utils import get_stock_data
|
|||||||
from screener.user_input import get_interval_choice, get_date_range
|
from screener.user_input import get_interval_choice, get_date_range
|
||||||
from indicators.three_atr_ema import ThreeATREMAIndicator
|
from indicators.three_atr_ema import ThreeATREMAIndicator
|
||||||
|
|
||||||
def check_entry_signal(df: pd.DataFrame) -> bool:
|
def check_entry_signal(df: pd.DataFrame) -> tuple:
|
||||||
"""
|
"""
|
||||||
Check for entry signal based on Three ATR EMA strategy
|
Check for entry signal based on Three ATR EMA strategy
|
||||||
|
|
||||||
@ -15,16 +15,16 @@ def check_entry_signal(df: pd.DataFrame) -> bool:
|
|||||||
df (pd.DataFrame): DataFrame with price data
|
df (pd.DataFrame): DataFrame with price data
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
bool: True if entry signal is present, False otherwise
|
tuple: (bool, datetime, dict) Entry signal, signal date, and signal data
|
||||||
"""
|
"""
|
||||||
if len(df) < 2: # Need at least 2 bars for comparison
|
if len(df) < 2: # Need at least 2 bars for comparison
|
||||||
return False
|
return False, None, None
|
||||||
|
|
||||||
indicator = ThreeATREMAIndicator()
|
indicator = ThreeATREMAIndicator()
|
||||||
results = indicator.calculate(df)
|
results = indicator.calculate(df)
|
||||||
|
|
||||||
if len(results) < 2:
|
if len(results) < 2:
|
||||||
return False
|
return False, None, None
|
||||||
|
|
||||||
# Get latest values
|
# Get latest values
|
||||||
current = df.iloc[-1]
|
current = df.iloc[-1]
|
||||||
@ -36,16 +36,19 @@ def check_entry_signal(df: pd.DataFrame) -> bool:
|
|||||||
prev_lower_band = results['lower_band'].iloc[-2]
|
prev_lower_band = results['lower_band'].iloc[-2]
|
||||||
|
|
||||||
# Entry conditions from Pine script:
|
# Entry conditions from Pine script:
|
||||||
# 1. Price is below EMA
|
|
||||||
# 2. Previous close was at or below lower band
|
|
||||||
# 3. Current close is higher than previous close
|
|
||||||
entry_signal = (
|
entry_signal = (
|
||||||
current['close'] < ema and
|
current['close'] < ema and
|
||||||
previous['close'] <= prev_lower_band and
|
previous['close'] <= prev_lower_band and
|
||||||
current['close'] > previous['close']
|
current['close'] > previous['close']
|
||||||
)
|
)
|
||||||
|
|
||||||
return entry_signal
|
signal_data = {
|
||||||
|
'price': current['close'],
|
||||||
|
'ema': ema,
|
||||||
|
'lower_band': lower_band
|
||||||
|
} if entry_signal else None
|
||||||
|
|
||||||
|
return entry_signal, current['date'] if entry_signal else None, signal_data
|
||||||
|
|
||||||
def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None:
|
def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None:
|
||||||
"""
|
"""
|
||||||
@ -118,16 +121,18 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int,
|
|||||||
if df.empty or len(df) < 21: # Need at least 21 bars for EMA
|
if df.empty or len(df) < 21: # Need at least 21 bars for EMA
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if check_entry_signal(df):
|
signal, signal_date, signal_data = check_entry_signal(df)
|
||||||
|
if signal:
|
||||||
signal_data = {
|
signal_data = {
|
||||||
'ticker': ticker,
|
'ticker': ticker,
|
||||||
'price': current_price,
|
'price': signal_data['price'],
|
||||||
'volume': current_volume,
|
'volume': current_volume,
|
||||||
|
'signal_date': signal_date,
|
||||||
'last_update': datetime.fromtimestamp(last_update/1000000000)
|
'last_update': datetime.fromtimestamp(last_update/1000000000)
|
||||||
}
|
}
|
||||||
|
|
||||||
if calculator:
|
if calculator:
|
||||||
position = calculator.calculate_position_size(current_price)
|
position = calculator.calculate_position_size(signal_data['price'])
|
||||||
signal_data.update({
|
signal_data.update({
|
||||||
'shares': position['shares'],
|
'shares': position['shares'],
|
||||||
'position_size': position['position_value'],
|
'position_size': position['position_value'],
|
||||||
@ -137,8 +142,8 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int,
|
|||||||
|
|
||||||
entry_signals.append(signal_data)
|
entry_signals.append(signal_data)
|
||||||
|
|
||||||
# Print signal information
|
# Print signal information with date
|
||||||
print(f"\n🔍 {ticker} @ ${current_price:.2f}")
|
print(f"\n🔍 {ticker} @ ${signal_data['price']:.2f} on {signal_date.strftime('%Y-%m-%d %H:%M')}")
|
||||||
if calculator:
|
if calculator:
|
||||||
print(f" Size: {signal_data['shares']} shares (${signal_data['position_size']:.2f})")
|
print(f" Size: {signal_data['shares']} shares (${signal_data['position_size']:.2f})")
|
||||||
print(f" Stop: ${signal_data['stop_loss']:.2f} (7%)")
|
print(f" Stop: ${signal_data['stop_loss']:.2f} (7%)")
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user