Python Screener: Pin Bar с аномальным объемом на 20 парах

import yfinance as yf
import pandas as pd
import numpy as np
import datetime

# --- Конфигурация скринера ---
SYMBOLS = [
    "AAPL", "MSFT", "GOOGL", "AMZN", "TSLA", "NVDA", "JPM", "V", "MA", "PG",
    "UNH", "HD", "KO", "PEP", "DIS", "NFLX", "ADBE", "CRM", "PYPL", "INTC"
] # Пример 20+ акций. Можно использовать 'BTC-USD', 'ETH-USD' для крипты или 'EURUSD=X' для форекса с yfinance.
TIMEFRAME = "1d" # Доступные таймфреймы: "1m", "2m", "5m", "15m", "30m", "60m", "90m", "1h", "1d", "5d", "1wk", "1mo", "3mo"
PERIOD = "60d" # Период для загрузки данных (например, "60d" для 60 дней)

# Параметры для определения Пин-бара
PIN_BAR_BODY_RATIO = 0.3  # Максимальный размер тела относительно общего диапазона свечи
PIN_BAR_WICK_RATIO = 0.6  # Минимальный размер длинного фитиля относительно общего диапазона свечи
PIN_BAR_NOSE_RATIO = 0.1  # Максимальный размер короткого фитиля относительно общего диапазона свечи

# Параметры для определения аномального объема
VOLUME_MA_PERIOD = 20     # Период для скользящего среднего объема
ANOMALOUS_VOLUME_MULTIPLIER = 1.5 # Множитель для определения аномального объема (например, 1.5x от MA)

# --- Функции для анализа ---

def fetch_ohlcv_data(symbol: str, period: str, interval: str) -> pd.DataFrame:
    """Загружает исторические OHLCV данные для заданного символа."""
    try:
        ticker = yf.Ticker(symbol)
        data = ticker.history(period=period, interval=interval)
        if data.empty:
            print(f"[WARNING] No data fetched for {symbol} with period {period} and interval {interval}")
            return pd.DataFrame()
        return data
    except Exception as e:
        print(f"[ERROR] Failed to fetch data for {symbol}: {e}")
        return pd.DataFrame()

def is_pin_bar(ohlc: pd.Series, body_ratio: float, wick_ratio: float, nose_ratio: float) -> bool:
    """Определяет, является ли свеча пин-баром."""
    open_price, high_price, low_price, close_price = ohlc['Open'], ohlc['High'], ohlc['Low'], ohlc['Close']

    total_range = high_price - low_price
    if total_range == 0: # Избегаем деления на ноль
        return False

    body_size = abs(close_price - open_price)
    upper_wick = high_price - max(open_price, close_price)
    lower_wick = min(open_price, close_price) - low_price

    # Критерий 1: Маленькое тело
    if body_size / total_range > body_ratio:
        return False

    # Критерий 2 & 3: Длинный хвост и короткий нос
    # Бычий пин-бар (длинный нижний фитиль)
    if lower_wick / total_range > wick_ratio and upper_wick / total_range < nose_ratio:
        # Дополнительное условие: длинный фитиль значительно больше тела
        if lower_wick > body_size * 2: # Например, в 2 раза больше тела
            return True

    # Медвежий пин-бар (длинный верхний фитиль)
    if upper_wick / total_range > wick_ratio and lower_wick / total_range < nose_ratio:
        # Дополнительное условие: длинный фитиль значительно больше тела
        if upper_wick > body_size * 2: # Например, в 2 раза больше тела
            return True

    return False

def is_anomalous_volume(df: pd.DataFrame, current_index: int, ma_period: int, multiplier: float) -> bool:
    """Определяет, является ли объем аномальным."""
    if current_index < ma_period:
        return False # Недостаточно данных для расчета MA

    # Убедимся, что 'Volume' существует и не пуст
    if 'Volume' not in df.columns or df['Volume'].isnull().all():
        return False

    # Рассчитываем скользящее среднее объема
    # Используем .iloc для доступа по целочисленному индексу
    rolling_volume_ma = df['Volume'].iloc[current_index - ma_period : current_index].mean()
    current_volume = df['Volume'].iloc[current_index]

    if rolling_volume_ma == 0:
        return False # Избегаем деления на ноль, если MA равно 0

    return current_volume > (rolling_volume_ma * multiplier)

# --- Основная логика скринера ---

def run_screener():
    print(f"Запуск скринера для {len(SYMBOLS)} символов на таймфрейме {TIMEFRAME}...")
    found_patterns = []

    for symbol in SYMBOLS:
        print(f"Обработка {symbol}...")
        df = fetch_ohlcv_data(symbol, PERIOD, TIMEFRAME)

        if df.empty or len(df) < VOLUME_MA_PERIOD + 1: # Нужен как минимум VOLUME_MA_PERIOD + 1 свеча для MA и текущей
            print(f"Пропуск {symbol}: недостаточно данных.")
            continue

        # Берем последнюю свечу для анализа
        last_candle_index = len(df) - 1
        last_candle = df.iloc[last_candle_index]

        # Проверяем на пин-бар
        is_pb = is_pin_bar(last_candle, PIN_BAR_BODY_RATIO, PIN_BAR_WICK_RATIO, PIN_BAR_NOSE_RATIO)

        # Проверяем на аномальный объем
        is_av = is_anomalous_volume(df, last_candle_index, VOLUME_MA_PERIOD, ANOMALOUS_VOLUME_MULTIPLIER)

        if is_pb and is_av:
            found_patterns.append({
                'symbol': symbol,
                'date': df.index[last_candle_index].strftime('%Y-%m-%d %H:%M'),
                'close': last_candle['Close'],
                'volume': last_candle['Volume'],
                'pattern': 'Pin Bar with Anomalous Volume'
            })
            print(f"  [НАЙДЕНО] {symbol} - Пин-бар с аномальным объемом на {df.index[last_candle_index].strftime('%Y-%m-%d %H:%M')}")

    if found_patterns:
        print("\n--- Результаты скрининга ---")
        for pattern in found_patterns:
            print(f"Символ: {pattern['symbol']}, Дата: {pattern['date']}, Закрытие: {pattern['close']:.2f}, Объем: {pattern['volume']:.0f}")
    else:
        print("\nПаттерны Пин-бар с аномальным объемом не найдены.")

if __name__ == "__main__":
    run_screener()