From aec5929ea7bff74cfce63c133f2b5924ef7dacb7 Mon Sep 17 00:00:00 2001 From: "Bobby (aider)" Date: Wed, 12 Feb 2025 22:09:39 -0800 Subject: [PATCH] feat: Add Heikin-Ashi scanner to technical scanner framework --- src/pages/screener/technical_scanner_page.py | 2 +- src/screener/scanner_controller.py | 4 +- src/screener/t_heikinashi.py | 119 +++++++++++++++++++ 3 files changed, 123 insertions(+), 2 deletions(-) create mode 100644 src/screener/t_heikinashi.py diff --git a/src/pages/screener/technical_scanner_page.py b/src/pages/screener/technical_scanner_page.py index 5f96bab..0108dd0 100644 --- a/src/pages/screener/technical_scanner_page.py +++ b/src/pages/screener/technical_scanner_page.py @@ -12,7 +12,7 @@ def technical_scanner_page(): with scanner_tab: scanner_type = st.selectbox( "Select Scanner", - ["SunnyBands", "ATR-EMA", "ATR-EMA v2"], + ["SunnyBands", "ATR-EMA", "ATR-EMA v2", "Heikin-Ashi"], key="tech_scanner_type" ) diff --git a/src/screener/scanner_controller.py b/src/screener/scanner_controller.py index 204ed24..4a1d5a8 100644 --- a/src/screener/scanner_controller.py +++ b/src/screener/scanner_controller.py @@ -2,6 +2,7 @@ from datetime import datetime from screener.t_sunnyband import run_sunny_scanner from screener.t_atr_ema import run_atr_ema_scanner from screener.t_atr_ema_v2 import run_atr_ema_scanner_v2 +from screener.t_heikinashi import run_heikin_ashi_scanner def run_technical_scanner(scanner_choice: str, start_date: str, end_date: str, min_price: float, max_price: float, min_volume: int, @@ -26,7 +27,8 @@ def run_technical_scanner(scanner_choice: str, start_date: str, end_date: str, scanner_map = { "sunnybands": lambda: run_sunny_scanner(min_price, max_price, min_volume, portfolio_size, interval, start_dt, end_dt), "atr-ema": lambda: run_atr_ema_scanner(min_price, max_price, min_volume, portfolio_size, interval, start_dt, end_dt), - "atr-ema_v2": lambda: run_atr_ema_scanner_v2(min_price, max_price, min_volume, portfolio_size, interval, start_dt, end_dt) + "atr-ema_v2": lambda: run_atr_ema_scanner_v2(min_price, max_price, min_volume, portfolio_size, interval, start_dt, end_dt), + "heikin-ashi": lambda: run_heikin_ashi_scanner(min_price, max_price, min_volume, portfolio_size, interval, start_dt, end_dt) } scanner_func = scanner_map.get(scanner_choice) diff --git a/src/screener/t_heikinashi.py b/src/screener/t_heikinashi.py new file mode 100644 index 0000000..18b9492 --- /dev/null +++ b/src/screener/t_heikinashi.py @@ -0,0 +1,119 @@ +import pandas as pd +from datetime import datetime, timedelta +from db.db_connection import create_client +from utils.data_utils import ( + get_stock_data, validate_signal_date, print_signal, + save_signals_to_csv, get_qualified_stocks +) +from utils.scanner_utils import initialize_scanner, process_signal_data +from trading.position_calculator import PositionCalculator + +def calculate_heikin_ashi(df: pd.DataFrame) -> pd.DataFrame: + """Calculate Heikin Ashi candles from regular OHLC data""" + ha_close = (df['open'] + df['high'] + df['low'] + df['close']) / 4 + ha_open = pd.Series(index=df.index) + ha_open.iloc[0] = df['open'].iloc[0] + for i in range(1, len(df)): + ha_open.iloc[i] = (ha_open.iloc[i-1] + ha_close.iloc[i-1]) / 2 + ha_high = df[['high', 'open', 'close']].max(axis=1) + ha_low = df[['low', 'open', 'close']].min(axis=1) + + return pd.DataFrame({ + 'ha_open': ha_open, + 'ha_high': ha_high, + 'ha_low': ha_low, + 'ha_close': ha_close + }) + +def check_entry_signal(df: pd.DataFrame) -> list: + """ + Check for bullish Heikin Ashi signals + + Args: + df (pd.DataFrame): DataFrame with price data + + Returns: + list: List of tuples (signal, date, signal_data) for each signal found + """ + if len(df) < 3: # Need at least 3 bars for comparison + return [] + + # Calculate Heikin Ashi values + ha_df = calculate_heikin_ashi(df) + signals = [] + + # Start from index 2 to compare with previous candles + for i in range(2, len(df)): + current = ha_df.iloc[i] + prev = ha_df.iloc[i-1] + prev2 = ha_df.iloc[i-2] + + # Bullish signal conditions: + # 1. Current candle is bullish (close > open) + # 2. Previous candle was bullish + # 3. Previous to previous candle was bearish (transition point) + if (current['ha_close'] > current['ha_open'] and + prev['ha_close'] > prev['ha_open'] and + prev2['ha_close'] < prev2['ha_open']): + + signal_data = { + 'price': df.iloc[i]['close'], + 'ha_open': current['ha_open'], + 'ha_close': current['ha_close'], + 'ha_high': current['ha_high'], + 'ha_low': current['ha_low'] + } + signals.append((True, df.iloc[i]['date'], signal_data)) + + return signals + +def run_heikin_ashi_scanner(min_price: float, max_price: float, min_volume: int, + portfolio_size: float = None, interval: str = "1d", + start_date: datetime = None, end_date: datetime = None) -> None: + """ + Run Heikin Ashi scanner to find bullish reversal patterns + """ + try: + # Initialize scanner components + interval, start_date, end_date, qualified_stocks, calculator = initialize_scanner( + min_price=min_price, + max_price=max_price, + min_volume=min_volume, + portfolio_size=portfolio_size, + interval=interval, + start_date=start_date, + end_date=end_date + ) + + if not qualified_stocks: + return + + bullish_signals = [] + + for ticker, current_price, current_volume, last_update, stock_type in qualified_stocks: + try: + df = get_stock_data(ticker, start_date, end_date, interval) + + if df.empty or len(df) < 3: # Need at least 3 bars + continue + + signals = check_entry_signal(df) + for signal, signal_date, signal_data in signals: + signal_data['date'] = signal_date + entry_data = process_signal_data( + ticker, signal_data, current_volume, + last_update, stock_type, calculator + ) + bullish_signals.append(entry_data) + print_signal(entry_data, "🟢") + + except Exception as e: + print(f"Error processing {ticker}: {str(e)}") + continue + + save_signals_to_csv(bullish_signals, 'heikin_ashi') + return bullish_signals + + except Exception as e: + print(f"Error during scan: {str(e)}") + return []