Backtest

# -*- coding: utf-8 -*-

import pandas as pd

import numpy as np

# --- ПАРАМЕТРЫ СТРАТЕГИИ И БЭКТЕСТА ---

# Эти параметры можно будет менять, чтобы улучшить результат

INITIAL_CAPITAL = 1000.0 # Начальный капитал в USDT для симуляции

RISK_PER_TRADE_PERCENT = 2.0 # Сколько % от капитала мы рискуем в одной сделке

RISK_REWARD_RATIO = 1.5 # Соотношение риска к прибыли (1.5 означает, что тейк-профит в 1.5 раза дальше стоп-лосса)

# Параметры для правил сигнала

PRICE_LOOKBACK_PERIOD = 5 # Сколько свечей назад смотрим для определения минимума

OI_LOOKBACK_PERIOD = 4  # Период для расчета среднего Открытого Интереса

MAX_FUNDING_RATE = 0.0001 # Максимальный фандинг для входа (0.01%)

# ---------------------------------------------------

def run_backtest():

print("--- Запуск исторического тестирования (Бэктеста) ---")

    # --- Шаг 1: Загрузка и подготовка данных ---

try:

    print("\n[1/4] Загрузка исторических данных из CSV файлов...")

        ohlcv_df = pd.read_csv('avax_ohlcv.csv')

    open_interest_df = pd.read_csv('AVAXUSDT_open_interest_15min.csv')

    funding_df = pd.read_csv('AVAXUSDT_funding_history.csv')

    # Приводим все временные метки к единому формату datetime

        ohlcv_df['timestamp'] = pd.to_datetime(ohlcv_df['timestamp'])

    open_interest_df['timestamp'] = pd.to_datetime(open_interest_df['timestamp'], utc=True)

    funding_df['fundingRateTimestamp'] = pd.to_datetime(funding_df['fundingRateTimestamp'], utc=True)

    # Агрегируем минутные свечи в 15-минутные

    ohlcv_df.set_index('timestamp', inplace=True)

    ohlcv_15m_df = ohlcv_df.resample('15T').agg({

        'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'

    }).dropna()

    ohlcv_15m_df.index = ohlcv_15m_df.index.tz_localize('UTC')

    # Готовим остальные данные

    open_interest_df.set_index('timestamp', inplace=True)

    funding_df.set_index('fundingRateTimestamp', inplace=True)

    funding_df = funding_df[~funding_df.index.duplicated(keep='first')]

    # Объединяем все в один датафрейм

    df = ohlcv_15m_df.join(open_interest_df['openInterest']).join(funding_df['fundingRate']).ffill().dropna()

    print("✅ Данные успешно загружены и объединены.")

except FileNotFoundError as e:

    print(f"ОШИБКА: Файл не найден! Убедитесь, что CSV файлы лежат в той же папке, что и скрипт.\n{e}")

    return

# --- Шаг 2: Расчет индикаторов и условий ---

print("\n[2/4] Расчет индикаторов и условий для стратегии...")

df['rolling_low'] = df['low'].shift(1).rolling(window=PRICE_LOOKBACK_PERIOD).min()

df['oi_average'] = df['openInterest'].shift(1).rolling(window=OI_LOOKBACK_PERIOD).mean()

# Условие 1: Сбор ликвидности и разворот

condition1 = (df['low'] < df['rolling_low']) & (df['close'] > df['open'])

# Условие 2: Рост Открытого Интереса

condition2 = df['openInterest'] > df['oi_average']

# Условие 3: Нейтральный или медвежий фандинг

condition3 = df['fundingRate'] <= MAX_FUNDING_RATE

df['signal'] = condition1 & condition2 & condition3

print("✅ Условия рассчитаны.")

    # --- Шаг 3: Симуляция торгов ---

print("\n[3/4] Запуск симуляции торгов... Это может занять некоторое время.")

trades = []

active_trade = None

    for i in range(len(df)):

    # Если у нас есть активная сделка, проверяем, не закрылась ли она

    if active_trade:

        # Проверка на Take Profit

        if df['high'].iloc[i] >= active_trade['take_profit']:

            active_trade['exit_price'] = active_trade['take_profit']

            active_trade['result'] = 'WIN'

            trades.append(active_trade)

            active_trade = None

            continue

        # Проверка на Stop Loss

        if df['low'].iloc[i] <= active_trade['stop_loss']:

      active_trade['exit_price'] = active_trade['stop_loss']

            active_trade['result'] = 'LOSS'

            trades.append(active_trade)

            active_trade = None

            continue

    # Если сигнала нет или уже есть активная сделка, пропускаем

    if not df['signal'].iloc[i] or active_trade:

        continue

    # Генерируем новую сделку

    entry_price = df['close'].iloc[i]

    stop_loss = df['low'].iloc[i]

        # Пропускаем сделку, если стоп-лосс слишком близко к цене входа

    if entry_price <= stop_loss:

        continue

        risk_per_unit = entry_price - stop_loss

    take_profit = entry_price + (risk_per_unit * RISK_REWARD_RATIO)

    active_trade = {

        'entry_date': df.index[i],

        'entry_price': entry_price,

        'stop_loss': stop_loss,

        'take_profit': take_profit,

        'result': None

    }

print("✅ Симуляция завершена.")

# --- Шаг 4: Анализ результатов ---

print("\n[4/4] Анализ результатов и формирование отчета...")

if not trades:

    print("\n--- ОТЧЕТ ПО БЭКТЕСТУ ---")

    print("За весь период не было сгенерировано ни одной сделки по заданным условиям.")

    return

wins = len([t for t in trades if t['result'] == 'WIN'])

losses = len([t for t in trades if t['result'] == 'LOSS'])

total_trades = len(trades)

win_rate = (wins / total_trades) * 100 if total_trades > 0 else 0

    # Симуляция изменения капитала

capital = INITIAL_CAPITAL

for trade in trades:

    risk_amount = capital * (RISK_PER_TRADE_PERCENT / 100)

    if trade['result'] == 'WIN':

        profit = risk_amount * RISK_REWARD_RATIO

        capital += profit

    elif trade['result'] == 'LOSS':

        capital -= risk_amount

print("\n\n--- 📈 ОТЧЕТ ПО БЭКТЕСТУ 📈 ---")

print(f"Период тестирования: с {df.index[0].date()} по {df.index[-1].date()}")

print("-------------------------------------------------")

print(f"Всего сделок: {total_trades}")

print(f"Прибыльных сделок (Wins): {wins}")

print(f"Убыточных сделок (Losses): {losses}")

    print(f"**Винрейт (Win Rate): {win_rate:.2f}%**")

print("-------------------------------------------------")

print(f"Начальный капитал: ${INITIAL_CAPITAL:,.2f}")

print(f"Конечный капитал: ${capital:,.2f}")

net_profit = capital - INITIAL_CAPITAL

net_profit_percent = (net_profit / INITIAL_CAPITAL) * 100

print(f"Чистая прибыль: ${net_profit:,.2f} ({net_profit_percent:.2f}%)")

print("-------------------------------------------------")

# --- Главный блок запуска ---

if name == "__main__":

run_backtest()