Strategy optimized for 5-second candles.
Timeframe
5s
Direction
Long Only
Stoploss
-2.0%
Trailing Stop
Yes
ROI
0m: 1.0%, 60m: 0.5%, 180m: 0.0%
Interface Version
3
Startup Candles
N/A
Indicators
5
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
#!/usr/bin/env python3
"""
Script to integrate custom candles with Freqtrade strategies.
This shows how to use custom candles in a strategy and backtest.
"""
import os
import sys
import logging
import argparse
from pathlib import Path
from typing import Dict, List, Optional, Any, Tuple
import pandas as pd
import numpy as np
# Add the parent directory to sys.path
sys.path.append(str(Path(__file__).parents[3]))
# Import freqtrade modules
from freqtrade.data.history.history_utils import pair_to_filename, load_pair_history
from freqtrade.data.converter import convert_ohlcv_format
from freqtrade.optimize.backtesting import Backtesting
from freqtrade.configuration import Configuration
from freqtrade.resolvers import StrategyResolver
from freqtrade.exchange import timeframe_to_minutes
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
# Constants
DEFAULT_EXCHANGE = "binance"
DEFAULT_TIMEFRAME = "5s"
BASE_DATA_DIR = Path("/allah/blue/ft/user_data/data")
CONFIG_FILE = Path("/allah/blue/ft/user_data/config/backtest.json")
class CustomCandleIntegrator:
"""Class to integrate custom candles with Freqtrade."""
def __init__(self, config_file: Path = CONFIG_FILE,
exchange_name: str = DEFAULT_EXCHANGE,
timeframe: str = DEFAULT_TIMEFRAME):
"""Initialize the integrator with config and timeframe."""
self.config_file = config_file
self.exchange_name = exchange_name
self.timeframe = timeframe
self.data_dir = BASE_DATA_DIR / exchange_name / timeframe
# Load configuration
try:
self.config = Configuration.from_files([str(config_file)])
# Update config with our settings
self.config['exchange']['name'] = exchange_name
self.config['timeframe'] = timeframe
self.config['datadir'] = str(BASE_DATA_DIR)
logger.info(f"Loaded configuration from {config_file}")
except Exception as e:
logger.error(f"Error loading config: {str(e)}")
raise
def load_custom_candles(self, pair: str) -> Optional[pd.DataFrame]:
"""Load custom candles for a specific pair."""
# Convert pair to filename format
filename = pair.replace('/', '-').replace(':', '')
file_path = self.data_dir / f"{filename}.parquet"
if not file_path.exists():
logger.warning(f"No data file found for {pair} at {file_path}")
return None
try:
# Load parquet file
df = pd.read_parquet(file_path)
# Ensure required columns exist
required_cols = ['date', 'open', 'high', 'low', 'close', 'volume']
if not all(col in df.columns for col in required_cols):
logger.error(f"Missing required columns in {file_path}. Found: {df.columns.tolist()}")
return None
# Convert to Freqtrade format
df_freqtrade = df[required_cols].copy()
# Make sure date is in milliseconds (int)
if isinstance(df_freqtrade['date'].iloc[0], pd.Timestamp):
df_freqtrade['date'] = df_freqtrade['date'].astype(np.int64) // 10**6
# Sort by date
df_freqtrade = df_freqtrade.sort_values('date')
logger.info(f"Loaded {len(df_freqtrade)} candles for {pair}")
return df_freqtrade
except Exception as e:
logger.error(f"Error loading data for {pair}: {str(e)}")
return None
def prepare_data_for_backtest(self, pairs: List[str]) -> Dict[str, pd.DataFrame]:
"""Prepare data for backtesting."""
data: Dict[str, pd.DataFrame] = {}
for pair in pairs:
df = self.load_custom_candles(pair)
if df is not None and not df.empty:
data[pair] = df
else:
logger.warning(f"Skipping {pair} due to missing or empty data")
if not data:
logger.error("No data available for backtesting")
return {}
return data
def run_backtest(self, strategy_name: str, pairs: List[str]) -> Optional[Dict[str, Any]]:
"""Run a backtest using the custom candles."""
# Prepare data
data = self.prepare_data_for_backtest(pairs)
if not data:
return None
# Configure backtesting
self.config['strategy'] = strategy_name
self.config['pairs'] = pairs
try:
# Initialize backtesting
backtesting = Backtesting(self.config)
backtesting.load_bt_data_detail = data # Set our custom data
backtesting.timeframe = self.timeframe
backtesting.timeframe_min = timeframe_to_minutes(self.timeframe)
# Load strategy
strategy = StrategyResolver.load_strategy(strategy_name, self.config)
# Run backtest
logger.info(f"Running backtest with strategy {strategy_name} on {len(pairs)} pairs")
results = backtesting.start(strategy)
return results
except Exception as e:
logger.error(f"Error during backtesting: {str(e)}")
return None
def display_backtest_results(self, results: Dict[str, Any]) -> None:
"""Display backtest results."""
if not results:
logger.error("No backtest results to display")
return
# Display summary
print(f"\n{'='*50}")
print(f"Backtest Results:")
print(f"{'='*50}")
# Extract key metrics
if 'results_per_pair' in results:
print("\nResults per pair:")
for pair, pair_results in results['results_per_pair'].items():
profit = pair_results.get('profit_abs', 0)
profit_pct = pair_results.get('profit_ratio', 0) * 100
trade_count = pair_results.get('trade_count', 0)
print(f" {pair}: {profit:.2f} ({profit_pct:.2f}%) over {trade_count} trades")
# Overall performance
if 'strategy_comparison' in results:
performance = results['strategy_comparison'][0]
total_profit = performance.get('profit_total_abs', 0)
total_profit_pct = performance.get('profit_total', 0) * 100
total_trades = performance.get('trade_count', 0)
win_rate = performance.get('win_ratio', 0) * 100
print(f"\nOverall performance:")
print(f" Total profit: {total_profit:.2f} ({total_profit_pct:.2f}%)")
print(f" Total trades: {total_trades}")
print(f" Win rate: {win_rate:.2f}%")
print(f"{'='*50}\n")
def create_custom_config(self, output_file: Path) -> None:
"""Create a custom configuration for using 5s candles."""
config = {
"max_open_trades": 1,
"stake_currency": "USDT",
"stake_amount": "unlimited",
"tradable_balance_ratio": 0.99,
"fiat_display_currency": "USD",
"timeframe": self.timeframe,
"dry_run": True,
"cancel_open_orders_on_exit": False,
"use_exit_signal": True,
"exit_profit_only": False,
"ignore_roi_if_entry_signal": False,
"exchange": {
"name": self.exchange_name,
"key": "",
"secret": "",
"ccxt_config": {},
"ccxt_async_config": {},
"pair_whitelist": [],
"pair_blacklist": []
},
"datadir": str(BASE_DATA_DIR),
"initial_state": "running",
"db_url": "sqlite:///user_data/tradesv3.sqlite",
"user_data_dir": "user_data",
"strategy": "SampleStrategy",
"strategy_path": "user_data/strategies/",
"internals": {
"process_throttle_secs": 5
}
}
# Write to file
try:
import json
with open(output_file, 'w') as f:
json.dump(config, f, indent=4)
logger.info(f"Created custom config at {output_file}")
except Exception as e:
logger.error(f"Error creating config file: {str(e)}")
def create_sample_strategy(self, output_file: Path) -> None:
"""Create a sample strategy optimized for 5s candles."""
strategy_code = """
from freqtrade.strategy import IStrategy, merge_informative_pair
from pandas import DataFrame
import talib.abstract as ta
import pandas as pd
import numpy as np
class FiveSecondStrategy(IStrategy):
"""
Strategy optimized for 5-second candles.
"""
# Strategy interface version
INTERFACE_VERSION = 3
# Minimal ROI designed for 5-second timeframe
minimal_roi = {
"0": 0.01, # 1% profit at any time
"60": 0.005, # 0.5% profit after 60 seconds (12 candles)
"180": 0 # 0% profit after 180 seconds (36 candles)
}
# Stoploss designed for 5-second timeframe
stoploss = -0.02 # 2% maximum loss
# Trailing stoploss (suitable for volatile 5s candles)
trailing_stop = True
trailing_stop_positive = 0.005
trailing_stop_positive_offset = 0.01
trailing_only_offset_is_reached = True
# Timeframe for the strategy
timeframe = '5s'
# Run "populate_indicators" only for new candle.
process_only_new_candles = True
# Number of candles the strategy requires before producing valid signals
startup_candle_count: int = 30
# Optional order type mapping
order_types = {
'entry': 'limit',
'exit': 'limit',
'stoploss': 'market',
'stoploss_on_exchange': False
}
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Adds several indicators to the given DataFrame.
Optimized for 5-second candles by using shorter windows.
"""
# Volume-weighted RSI (shorter windows for 5s candles)
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=9)
# EMA - using short windows
dataframe['ema5'] = ta.EMA(dataframe, timeperiod=5)
dataframe['ema10'] = ta.EMA(dataframe, timeperiod=10)
dataframe['ema20'] = ta.EMA(dataframe, timeperiod=20)
# Bollinger Bands - adjusted for 5s volatility
bollinger = ta.BBANDS(dataframe, timeperiod=12, nbdevup=2.0, nbdevdn=2.0)
dataframe['bb_lowerband'] = bollinger['lowerband']
dataframe['bb_middleband'] = bollinger['middleband']
dataframe['bb_upperband'] = bollinger['upperband']
# MACD - faster settings for 5s
macd = ta.MACD(dataframe, fastperiod=6, slowperiod=12, signalperiod=3)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macdhist'] = macd['macdhist']
# Volume moving average - shorter window
dataframe['volume_mean'] = dataframe['volume'].rolling(6).mean()
# Volatility indicator
dataframe['atr'] = ta.ATR(dataframe, timeperiod=7)
# Price change rate over short periods (suitable for 5s)
dataframe['pct_change'] = dataframe['close'].pct_change(3)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, defines entry signals.
Optimized for 5-second candles by looking for quick momentum shifts.
"""
dataframe.loc[
(
# MACD rising and crossing signal
(dataframe['macd'] > dataframe['macdsignal']) &
(dataframe['macd'].shift(1) < dataframe['macdsignal'].shift(1)) &
# RSI shows momentum but not overbought
(dataframe['rsi'] > 45) & (dataframe['rsi'] < 70) &
# EMA alignment shows uptrend
(dataframe['ema5'] > dataframe['ema10']) &
(dataframe['ema10'] > dataframe['ema20']) &
# Volume is increasing
(dataframe['volume'] > dataframe['volume_mean']) &
# Volatility is reasonable
(dataframe['atr'] < dataframe['close'] * 0.01) &
# Recent positive momentum
(dataframe['pct_change'] > 0.001)
),
'enter_long'] = 1
dataframe.loc[
(
# MACD falling and crossing signal
(dataframe['macd'] < dataframe['macdsignal']) &
(dataframe['macd'].shift(1) > dataframe['macdsignal'].shift(1)) &
# RSI shows downward momentum but not oversold
(dataframe['rsi'] < 55) & (dataframe['rsi'] > 30) &
# EMA alignment shows downtrend
(dataframe['ema5'] < dataframe['ema10']) &
(dataframe['ema10'] < dataframe['ema20']) &
# Volume is increasing
(dataframe['volume'] > dataframe['volume_mean']) &
# Volatility is reasonable
(dataframe['atr'] < dataframe['close'] * 0.01) &
# Recent negative momentum
(dataframe['pct_change'] < -0.001)
),
'enter_short'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on TA indicators, defines exit signals.
Optimized for 5-second candles by using quick reversal signals.
"""
dataframe.loc[
(
# MACD crossing signal to the downside
((dataframe['macd'] < dataframe['macdsignal']) &
(dataframe['macd'].shift(1) > dataframe['macdsignal'].shift(1))) |
# RSI overbought
(dataframe['rsi'] > 75) |
# Price hit upper Bollinger Band
(dataframe['close'] > dataframe['bb_upperband']) |
# EMAs showing trend weakness
(dataframe['ema5'] < dataframe['ema10'])
),
'exit_long'] = 1
dataframe.loc[
(
# MACD crossing signal to the upside
((dataframe['macd'] > dataframe['macdsignal']) &
(dataframe['macd'].shift(1) < dataframe['macdsignal'].shift(1))) |
# RSI oversold
(dataframe['rsi'] < 25) |
# Price hit lower Bollinger Band
(dataframe['close'] < dataframe['bb_lowerband']) |
# EMAs showing trend weakness
(dataframe['ema5'] > dataframe['ema10'])
),
'exit_short'] = 1
return dataframe
"""
# Write to file
try:
with open(output_file, 'w') as f:
f.write(strategy_code.strip())
logger.info(f"Created sample strategy at {output_file}")
except Exception as e:
logger.error(f"Error creating strategy file: {str(e)}")
def main():
parser = argparse.ArgumentParser(description='Integrate custom candles with Freqtrade')
parser.add_argument('--config', type=str, default=str(CONFIG_FILE),
help=f'Configuration file path (default: {CONFIG_FILE})')
parser.add_argument('--exchange', type=str, default=DEFAULT_EXCHANGE,
help=f'Exchange name (default: {DEFAULT_EXCHANGE})')
parser.add_argument('--timeframe', type=str, default=DEFAULT_TIMEFRAME,
help=f'Candle timeframe (default: {DEFAULT_TIMEFRAME})')
parser.add_argument('--pairs', type=str, nargs='+', default=["ETH/USDT:USDT"],
help='Trading pairs to use (default: ["ETH/USDT:USDT"])')
parser.add_argument('--strategy', type=str, default="FiveSecondStrategy",
help='Strategy name to use for backtesting')
parser.add_argument('--create-config', action='store_true',
help='Create a custom config file for 5s candles')
parser.add_argument('--create-strategy', action='store_true',
help='Create a sample strategy optimized for 5s candles')
parser.add_argument('--backtest', action='store_true',
help='Run a backtest with the custom candles')
args = parser.parse_args()
# Initialize integrator
integrator = CustomCandleIntegrator(
Path(args.config), args.exchange, args.timeframe
)
# Create config if requested
if args.create_config:
integrator.create_custom_config(Path(args.config))
# Create strategy if requested
if args.create_strategy:
strategy_dir = Path('/allah/blue/ft/user_data/strategies')
strategy_dir.mkdir(parents=True, exist_ok=True)
strategy_path = strategy_dir / f"{args.strategy}.py"
integrator.create_sample_strategy(strategy_path)
# Run backtest if requested
if args.backtest:
results = integrator.run_backtest(args.strategy, args.pairs)
if results:
integrator.display_backtest_results(results)
return 0
if __name__ == "__main__":
sys.exit(main())