bot/data/market_structure.py - dev - fgtmenow/tb

bot/data/market_structure.py
482 строки23 KB
Azarev Artem
order placement
19 апр 2025, 01:38
19 апр 2025, 01:38cd7faee
import numpy as np
import pandas as pd


def find_initial_swing_points(df: pd.DataFrame, window: int) -> None:
    """
    Находит начальные точки разворота (swing points) на графике цены.

    Args:
        df: DataFrame с данными OHLCV, должен содержать колонки 'high' и 'low'
        window: Окно поиска минимумов и максимумов

    Добавляет колонки в DataFrame:
        - swing_point: 1 для максимумов, -1 для минимумов
        - swing_level: Уровни цен точек разворота
    """
    # Находим начальные точки разворота
    swing_highs_lows: np.ndarray = np.where(
        df["high"] == df["high"].shift(-(window // 2)).rolling(window).max(),
        1,  # максимум
        np.where(
            df["low"] == df["low"].shift(-(window // 2)).rolling(window).min(),
            -1,  # минимум
            np.nan  # не точка разворота
        ),
    )

    # Обработка краевых случаев
    positions: np.ndarray = np.where(~np.isnan(swing_highs_lows))[0]
    if len(positions) > 0:
        # Если первая точка максимум - добавляем минимум в начало
        if swing_highs_lows[positions[0]] == 1:
            swing_highs_lows[0] = -1
        # Если первая точка минимум - добавляем максимум в начало
        if swing_highs_lows[positions[0]] == -1:
            swing_highs_lows[0] = 1
        # Если последняя точка минимум - добавляем максимум в конец
        if swing_highs_lows[positions[-1]] == -1:
            swing_highs_lows[-1] = 1
        # Если последняя точка максимум - добавляем минимум в конец
        if swing_highs_lows[positions[-1]] == 1:
            swing_highs_lows[-1] = -1

    # Вычисляем уровень
    level: np.ndarray = np.where(
        swing_highs_lows == 1,
        df["high"],
        np.where(swing_highs_lows == -1, df["low"], np.nan)
    )

    # Добавляем результаты в DataFrame
    df["swing_point"] = swing_highs_lows
    df["swing_level"] = level
    df["swing_retracement"] = np.nan  # Инициализируем столбец для ретрейсментов


def filter_swing_points_by_fibo(df: pd.DataFrame, fibo_level: float) -> None:
    """
    Фильтрует точки разворота по уровню Фибоначчи.

    Args:
        df: DataFrame с данными OHLCV и найденными точками разворота
        fibo_level: Уровень фибоначи для фильтрации истинной коррекции

    Добавляет/обновляет колонку в DataFrame:
        - swing_retracement: Процент отката для каждой точки
    """
    if "swing_point" not in df.columns:
        raise ValueError("Сначала должна быть вызвана функция find_initial_swing_points")

    # Используем отдельный массив numpy для манипуляций с точками разворота
    # Убедимся, что массив имеет тип float для поддержки NaN
    swing_points_array: np.ndarray = df["swing_point"].values.copy().astype(float)

    # Инициализируем массив для ретрейсментов
    retracements = np.full_like(swing_points_array, np.nan, dtype=float)

    # Основной цикл - он пока необходим, т.к. фильтрация меняет структуру массива
    while True:
        # Находим позиции всех непустых точек разворота
        positions = np.where(~np.isnan(swing_points_array))[0]
        if len(positions) < 4:  # Нужно минимум 4 точки для анализа
            break

        # Создаем массивы для P1, P2, P3, P4 позиций и значений
        p_positions = np.column_stack([
            positions[:-3],  # P1 позиции
            positions[1:-2],  # P2 позиции
            positions[2:-1],  # P3 позиции
            positions[3:]  # P4 позиции
        ])

        # Получаем массив уровней для всех точек разворота
        levels = np.where(
            swing_points_array == 1,
            df["high"].values,
            np.where(swing_points_array == -1, df["low"].values, np.nan)
        )

        # Векторизованный расчет ретрейсментов для 3-й и 4-й точек
        found_invalid = False

        for i, p_pos in enumerate(p_positions):
            p1_pos, p2_pos, p3_pos, p4_pos = map(int, p_pos)  # Явно преобразуем в int

            # Расчет ретрейсмента для 4-й точки
            # Явно преобразуем элементы numpy-массива в float
            level2: float = float(levels[p2_pos])
            level3: float = float(levels[p3_pos])
            level4: float = float(levels[p4_pos])

            impulse_size = abs(level3 - level2)
            if impulse_size > 0:
                p2p4_retracement = abs(level4 - level3) / impulse_size * 100
            else:
                p2p4_retracement = np.nan

            if not np.isnan(p2p4_retracement):
                df.iloc[p4_pos, df.columns.get_loc("swing_retracement")] = p2p4_retracement
                retracements[p4_pos] = p2p4_retracement

            # Проверка паттернов
            # Используем типы int для получения точного порядка
            order = [
                int(swing_points_array[p1_pos]),
                int(swing_points_array[p2_pos]),
                int(swing_points_array[p3_pos]),
                int(swing_points_array[p4_pos])
            ]

            # Проверяем паттерны и фильтруем по уровню Фибоначчи
            if (order == [1, -1, 1, -1] and not np.isnan(p2p4_retracement) and
                    p2p4_retracement < fibo_level):
                swing_points_array[p4_pos] = np.nan
                found_invalid = True
                # Обновляем массив точек разворота в DataFrame временно для очистки
                df_temp = df.copy()
                df_temp["swing_point"] = swing_points_array
                cleanup_consecutive(df_temp, df_temp["swing_point"].values)
                swing_points_array = df_temp["swing_point"].values.astype(
                    float)  # Преобразуем в float для поддержки NaN
                break

            if (order == [-1, 1, -1, 1] and not np.isnan(p2p4_retracement) and
                    p2p4_retracement < fibo_level):
                swing_points_array[p4_pos] = np.nan
                found_invalid = True
                # Обновляем массив точек разворота в DataFrame временно для очистки
                df_temp = df.copy()
                df_temp["swing_point"] = swing_points_array
                cleanup_consecutive(df_temp, df_temp["swing_point"].values)
                swing_points_array = df_temp["swing_point"].values.astype(
                    float)  # Преобразуем в float для поддержки NaN
                break

        # Если не найдено недопустимых точек, выходим из цикла
        if not found_invalid:
            break

    # Обновляем DataFrame отфильтрованными данными
    df["swing_point"] = swing_points_array


def resample_ohlcv_data(df: pd.DataFrame, resample_period: str) -> pd.DataFrame:
    """
    Ресемплирует данные OHLCV на указанный период, учитывая только полные периоды.

    Args:
        df: DataFrame с данными OHLCV и временным индексом
        resample_period: Период ресемплирования данных (например, '5min', '1H', '4H')

    Returns:
        pd.DataFrame: Ресемплированный DataFrame
    """
    # Преобразуем период в timedelta для вычислений
    period = pd.Timedelta(resample_period)

    # Находим начало и конец полных периодов
    start_time = df.index[0].ceil(period)
    end_time = df.index[-1].floor(period)

    # Создаем новый DataFrame только с полными периодами
    df_filtered = df[(df.index >= start_time) & (df.index < end_time)]

    # Ресемплируем данные
    df_resampled = df_filtered.resample(resample_period).agg({
        'open': 'first',
        'high': 'max',
        'low': 'min',
        'close': 'last',
        'volume': 'sum',
        'timestamp': 'first'
    }).dropna()

    # Обновляем timestamp после ресемплирования
    df_resampled['timestamp'] = df_resampled.index.astype(int) // 10 ** 6
    df_resampled.index = pd.to_datetime(df_resampled['timestamp'], unit='ms')
    df_resampled = df_resampled.sort_index()
    return df_resampled


def _find_swing_points_internal(df: pd.DataFrame, fibo_level: float, window: int) -> None:
    """Internal function that implements the original find_swing_points logic"""
    # Находим начальные точки разворота
    swing_highs_lows: np.ndarray = np.where(
        df["high"] == df["high"].shift(-(window // 2)).rolling(window).max(),
        1,  # максимум
        np.where(
            df["low"] == df["low"].shift(-(window // 2)).rolling(window).min(),
            -1,  # минимум
            np.nan  # не точка разворота
        ),
    )
    cleanup_consecutive(df, swing_highs_lows)

    # Обработка краевых случаев
    positions: np.ndarray = np.where(~np.isnan(swing_highs_lows))[0]
    if len(positions) > 0:
        # Если первая точка максимум - добавляем минимум в начало
        if swing_highs_lows[positions[0]] == 1:
            swing_highs_lows[0] = -1
        # Если первая точка минимум - добавляем максимум в начало
        if swing_highs_lows[positions[0]] == -1:
            swing_highs_lows[0] = 1
        # Если последняя точка минимум - добавляем максимум в конец
        # if swing_highs_lows[positions[-1]] == -1:
        #    swing_highs_lows[-1] = 1
        # Если последняя точка максимум - добавляем минимум в конец
        # if swing_highs_lows[positions[-1]] == 1:
        #    swing_highs_lows[-1] = -1

    # Вычисляем уровень
    level: np.ndarray = np.where(
        swing_highs_lows == 1,
        df["high"],
        np.where(swing_highs_lows == -1, df["low"], np.nan)
    )

    # Добавляем результаты в оригинальный DataFrame
    df["swing_point"] = swing_highs_lows
    df["swing_level"] = level
    df["swing_retracement"] = np.nan


def cleanup_consecutive(df: pd.DataFrame, swing_highs_lows: np.ndarray) -> None:
    """
    Оптимизированная версия функции очистки последовательных точек разворота.
    Использует векторизованные операции вместо циклов.
    """
    while True:
        # Получаем позиции непустых точек разворота
        positions = np.where(~np.isnan(swing_highs_lows))[0]
        if len(positions) < 2:
            break

        # Векторизованное сравнение соседних точек
        current = swing_highs_lows[positions[:-1]]
        next_points = swing_highs_lows[positions[1:]]

        # Проверяем последовательные точки одного типа
        consecutive_mask = (current == next_points)
        if not consecutive_mask.any():
            break

        # Более эффективная обработка уровней
        current_levels = np.where(
            swing_highs_lows[positions[:-1]] == 1,
            df["high"].iloc[positions[:-1]],
            df["low"].iloc[positions[:-1]]
        )

        next_levels = np.where(
            swing_highs_lows[positions[1:]] == 1,
            df["high"].iloc[positions[1:]],
            df["low"].iloc[positions[1:]]
        )

        # Создаем маску для удаления точек более эффективно
        index_to_remove = np.zeros(len(positions), dtype=bool)

        # Обработка последовательных точек одного типа
        consecutive_same = (current == next_points)
        is_highs = (current == 1)

        # Стандартная логика фильтрации для оставшихся точек
        index_to_remove[:-1] |= consecutive_same & (
                (is_highs & (current_levels < next_levels)) |
                (~is_highs & (current_levels > next_levels))
        )
        index_to_remove[1:] |= consecutive_same & (
                (is_highs & (current_levels >= next_levels)) |
                (~is_highs & (current_levels <= next_levels))
        )

        if not index_to_remove.any():
            break

        swing_highs_lows[positions[index_to_remove]] = np.nan


def market_structure(df: pd.DataFrame) -> None:
    """
    Анализирует структуру рынка на основе точек разворота (swing points).

    Определяет:
    - BOS (Break of Structure): когда цена пробивает предыдущий структурный уровень
    - CHoCH (Change of Character): первый BOS в новом направлении (смена тренда)
    - Confirm: второй BOS в том же направлении (подтверждение тренда)

    Args:
        df: DataFrame с данными OHLCV и определенными точками разворота
    """
    # Проверяем и инициализируем индекс, если он не определен
    if df.index.name is None and len(df.index) > 0:
        df.index.name = 'index'

    # Инициализируем столбцы - используем loc чтобы избежать SettingWithCopyWarning
    df['bos'] = np.nan
    df['choch'] = np.nan
    df['confirm'] = np.nan
    df['trend'] = np.nan
    df['struct_level'] = np.nan
    df['struct_direction'] = np.nan

    # Преобразуем struct_start_idx и struct_broken_idx в тип object для правильного хранения временных меток
    df['struct_start_idx'] = pd.Series(dtype='object')
    df['struct_broken_idx'] = pd.Series(dtype='object')

    # Получаем все точки разворота
    swing_points = df[~df['swing_point'].isna()]
    if len(swing_points) < 3:
        return  # Необходимо минимум 3 точки разворота для анализа структуры

    last_direction = 0  # 0: нет направления, 1: вверх, -1: вниз
    bos_count = 0  # Счетчик BOS в текущем направлении
    first_bos_found = False  # Отслеживаем, нашли ли мы первый BOS

    # Перебираем точки разворота, начиная с третьей точки
    for i in range(3, len(swing_points)):
        current_idx = swing_points.index[i]
        prev_prev_idx = swing_points.index[i - 2]

        current_point = swing_points.loc[current_idx]
        prev_prev_point = swing_points.loc[prev_prev_idx]

        # BOS ВВЕРХ: текущий максимум пробивает предыдущий максимум
        if (current_point['swing_point'] == 1 and
                prev_prev_point['swing_point'] == 1 and
                current_point['high'] > prev_prev_point['high']):

            direction = 1
            struct_level = prev_prev_point['high']
            struct_start_idx = prev_prev_idx

            # Находим свечи, где цена закрытия пробивает уровень
            candles_between = df.loc[prev_prev_idx:current_idx]
            potential_breaks = candles_between[candles_between['close'] > struct_level]

            struct_broken_idx = None
            if not potential_breaks.empty:
                for j in range(len(potential_breaks) - 1):  # Проверяем все потенциальные свечи пробоя, кроме последней
                    break_idx = potential_breaks.index[j]
                    next_idx_loc = df.index.get_loc(break_idx) + 1
                    if next_idx_loc < len(df.index):
                        next_idx = df.index[next_idx_loc]
                        # Подтверждаем, что тело следующей свечи полностью выше уровня
                        if df.loc[next_idx, 'open'] > struct_level and df.loc[next_idx, 'close'] > struct_level:
                            struct_broken_idx = break_idx
                            break

            # Если подтвержденный пробой не найден, используем текущую точку разворота
            if struct_broken_idx is None:
                struct_broken_idx = current_idx

            # Сохраняем общие данные о структуре
            df.loc[current_idx, 'struct_level'] = struct_level
            df.loc[current_idx, 'struct_direction'] = direction
            df.loc[current_idx, 'struct_start_idx'] = struct_start_idx
            df.loc[current_idx, 'struct_broken_idx'] = struct_broken_idx

            # Первый BOS всегда CHoCH
            if not first_bos_found:
                df.loc[current_idx, 'choch'] = 1
                bos_count = 1
                first_bos_found = True
            # Проверяем, является ли это CHoCH (изменение направления)
            elif last_direction == -1:
                df.loc[current_idx, 'choch'] = 1
                bos_count = 1
            # Проверяем, является ли это подтверждением (второй BOS в том же направлении)
            elif last_direction == 1 and bos_count == 1:
                df.loc[current_idx, 'confirm'] = 1
                bos_count = 2
            # Иначе это обычный BOS
            else:
                df.loc[current_idx, 'bos'] = 1
                bos_count += 1

            last_direction = direction

        # BOS ВНИЗ: текущий минимум пробивает предыдущий минимум
        elif (current_point['swing_point'] == -1 and
              prev_prev_point['swing_point'] == -1 and
              current_point['low'] < prev_prev_point['low']):

            direction = -1
            struct_level = prev_prev_point['low']
            struct_start_idx = prev_prev_idx

            # Находим свечи, где цена закрытия пробивает уровень
            candles_between = df.loc[prev_prev_idx:current_idx]
            potential_breaks = candles_between[candles_between['close'] < struct_level]

            struct_broken_idx = None
            if not potential_breaks.empty:
                for j in range(len(potential_breaks) - 1):  # Проверяем все потенциальные свечи пробоя, кроме последней
                    break_idx = potential_breaks.index[j]
                    next_idx_loc = df.index.get_loc(break_idx) + 1
                    if next_idx_loc < len(df.index):
                        next_idx = df.index[next_idx_loc]
                        # Подтверждаем, что тело следующей свечи полностью ниже уровня
                        if df.loc[next_idx, 'open'] < struct_level and df.loc[next_idx, 'close'] < struct_level:
                            struct_broken_idx = break_idx
                            break

            # Если подтвержденный пробой не найден, используем текущую точку разворота
            if struct_broken_idx is None:
                struct_broken_idx = current_idx

            # Сохраняем общие данные о структуре
            df.loc[current_idx, 'struct_level'] = struct_level
            df.loc[current_idx, 'struct_direction'] = direction
            df.loc[current_idx, 'struct_start_idx'] = struct_start_idx
            df.loc[current_idx, 'struct_broken_idx'] = struct_broken_idx

            # Первый BOS всегда CHoCH
            if not first_bos_found:
                df.loc[current_idx, 'choch'] = 1
                bos_count = 1
                first_bos_found = True
            # Проверяем, является ли это CHoCH (изменение направления)
            elif last_direction == 1:
                df.loc[current_idx, 'choch'] = 1
                bos_count = 1
            # Проверяем, является ли это подтверждением (второй BOS в том же направлении)
            elif last_direction == -1 and bos_count == 1:
                df.loc[current_idx, 'confirm'] = 1
                bos_count = 2
            # Иначе это обычный BOS
            else:
                df.loc[current_idx, 'bos'] = 1
                bos_count += 1

            last_direction = direction

    # Отмечаем диапазоны тренда
    if last_direction != 0:
        # Находим все индексы, где у нас есть подтвержденные тренды
        confirm_points = df[~df['confirm'].isna()]
        choch_points = df[~df['choch'].isna()]

        if not confirm_points.empty:
            for i in range(len(confirm_points)):
                confirm_idx = confirm_points.index[i]

                # Получаем фактическую свечу, которая пробила структуру
                struct_broken_idx = df.loc[confirm_idx, 'struct_broken_idx']

                # Находим следующий CHoCH после этого подтверждения
                next_choch = choch_points[choch_points.index > confirm_idx]

                if not next_choch.empty:
                    next_choch_idx = next_choch.index[0]
                    # Получаем фактическую свечу, которая пробила структуру для CHoCH
                    next_choch_broken_idx = df.loc[next_choch_idx, 'struct_broken_idx']
                    end_idx = next_choch_broken_idx
                else:
                    end_idx = df.index[-1]

                # Отмечаем тренд для этого диапазона - начиная со свечи пробоя структуры
                direction = df.loc[confirm_idx, 'struct_direction']
                df.loc[struct_broken_idx:end_idx, 'trend'] = direction