Simon/Cycles
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor to remove rebundant parameters and use just a config file by default too

Browse Source
This commit is contained in:
Vasily.onl 2025-05-23 17:55:13 +08:00
parent 256ad67742
commit ebd8ef3d87
5 changed files with 176 additions and 224 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
configs/config_bbrs.json
View File

@ -3,7 +3,6 @@
"stop_date": "2025-03-15", "stop_date": "2025-03-15",
"initial_usd": 10000, "initial_usd": 10000,
"timeframes": ["1min"], "timeframes": ["1min"],
"stop_loss_pcts": [0.05],
"strategies": [ "strategies": [
{ {
"name": "bbrs", "name": "bbrs",
@ -17,7 +16,8 @@
"sideways_rsi_threshold": [40, 60], "sideways_rsi_threshold": [40, 60],
"sideways_bb_multiplier": 1.8, "sideways_bb_multiplier": 1.8,
"strategy_name": "MarketRegimeStrategy", "strategy_name": "MarketRegimeStrategy",
"SqueezeStrategy": true "SqueezeStrategy": true,
"stop_loss_pct": 0.05
} }
} }
], ],

14
configs/config_combined.json
View File

@ -3,12 +3,14 @@
"stop_date": "2025-03-15", "stop_date": "2025-03-15",
"initial_usd": 10000, "initial_usd": 10000,
"timeframes": ["15min"], "timeframes": ["15min"],
"stop_loss_pcts": [0.04],
"strategies": [ "strategies": [
{ {
"name": "default", "name": "default",
"weight": 0.6, "weight": 0.6,
"params": {} "params": {
"timeframe": "15min",
"stop_loss_pct": 0.03
}
}, },
{ {
"name": "bbrs", "name": "bbrs",
@ -17,7 +19,13 @@
"bb_width": 0.05, "bb_width": 0.05,
"bb_period": 20, "bb_period": 20,
"rsi_period": 14, "rsi_period": 14,
"strategy_name": "MarketRegimeStrategy" "trending_rsi_threshold": [30, 70],
"trending_bb_multiplier": 2.5,
"sideways_rsi_threshold": [40, 60],
"sideways_bb_multiplier": 1.8,
"strategy_name": "MarketRegimeStrategy",
"SqueezeStrategy": true,
"stop_loss_pct": 0.05
} }
} }
], ],

6
configs/config_default.json
View File

@ -3,12 +3,14 @@
"stop_date": "2025-03-15", "stop_date": "2025-03-15",
"initial_usd": 10000, "initial_usd": 10000,
"timeframes": ["15min"], "timeframes": ["15min"],
"stop_loss_pcts": [0.03, 0.05],
"strategies": [ "strategies": [
{ {
"name": "default", "name": "default",
"weight": 1.0, "weight": 1.0,
"params": {} "params": {
"timeframe": "15min",
"stop_loss_pct": 0.03
}
} }
], ],
"combination_rules": { "combination_rules": {

4
configs/config_default_5min.json
View File

@ -3,13 +3,13 @@
"stop_date": "2024-01-31", "stop_date": "2024-01-31",
"initial_usd": 10000, "initial_usd": 10000,
"timeframes": ["5min"], "timeframes": ["5min"],
"stop_loss_pcts": [0.03, 0.05],
"strategies": [ "strategies": [
{ {
"name": "default", "name": "default",
"weight": 1.0, "weight": 1.0,
"params": { "params": {
"timeframe": "5min" "timeframe": "5min",
"stop_loss_pct": 0.03
} }
} }
], ],

372
main.py
View File

@ -38,183 +38,169 @@ def strategy_manager_exit(backtester: Backtest, df_index: int):
"""Strategy Manager exit function""" """Strategy Manager exit function"""
return backtester.strategy_manager.get_exit_signal(backtester, df_index) return backtester.strategy_manager.get_exit_signal(backtester, df_index)
def process_timeframe_data(min1_df, df, stop_loss_pcts, rule_name, initial_usd, strategy_config=None, debug=False): def process_timeframe_data(data_1min, timeframe, config, debug=False):
"""Process the entire timeframe with all stop loss values using Strategy Manager""" """Process a timeframe using Strategy Manager with configuration"""
results_rows = [] results_rows = []
trade_rows = [] trade_rows = []
for stop_loss_pct in stop_loss_pcts: # Extract values from config
# Create and initialize strategy manager initial_usd = config['initial_usd']
if strategy_config: strategy_config = {
# Use provided strategy configuration "strategies": config['strategies'],
strategy_manager = create_strategy_manager(strategy_config) "combination_rules": config['combination_rules']
else: }
# Default to single default strategy for backward compatibility
default_strategy_config = {
"strategies": [
{
"name": "default",
"weight": 1.0,
"params": {"stop_loss_pct": stop_loss_pct}
}
],
"combination_rules": {
"entry": "any",
"exit": "any",
"min_confidence": 0.5
}
}
strategy_manager = create_strategy_manager(default_strategy_config)
# Inject stop_loss_pct into all strategy params if not present
for strategy in strategy_manager.strategies:
if "stop_loss_pct" not in strategy.params:
strategy.params["stop_loss_pct"] = stop_loss_pct
# Get the primary timeframe from the first strategy for backtester setup
primary_strategy = strategy_manager.strategies[0]
primary_timeframe = primary_strategy.get_timeframes()[0]
# For BBRS strategy, it works with 1-minute data directly and handles internal resampling
# For other strategies, use their preferred timeframe
if primary_strategy.name == "bbrs":
# BBRS strategy processes 1-minute data and outputs signals on its internal timeframes
# Use 1-minute data for backtester working dataframe
working_df = min1_df.copy()
else:
# Other strategies specify their preferred timeframe
# Create backtester working data from the primary strategy's primary timeframe
temp_backtester = type('temp', (), {})()
temp_backtester.original_df = min1_df
# Let the primary strategy resample the data to get the working dataframe
primary_strategy._resample_data(min1_df)
working_df = primary_strategy.get_primary_timeframe_data()
# Prepare working dataframe for backtester (ensure timestamp column)
working_df_for_backtest = working_df.copy().reset_index()
if 'index' in working_df_for_backtest.columns:
working_df_for_backtest = working_df_for_backtest.rename(columns={'index': 'timestamp'})
# Initialize backtest with strategy manager initialization
backtester = Backtest(initial_usd, working_df_for_backtest, working_df_for_backtest, strategy_manager_init)
# Store original min1_df for strategy processing
backtester.original_df = min1_df
# Attach strategy manager to backtester and initialize
backtester.strategy_manager = strategy_manager
strategy_manager.initialize(backtester)
# Run backtest with strategy manager functions # Create and initialize strategy manager
results = backtester.run( if not strategy_config:
strategy_manager_entry, logging.error("No strategy configuration provided")
strategy_manager_exit, return results_rows, trade_rows
debug
) strategy_manager = create_strategy_manager(strategy_config)
# Get the primary timeframe from the first strategy for backtester setup
primary_strategy = strategy_manager.strategies[0]
primary_timeframe = primary_strategy.get_timeframes()[0]
# For BBRS strategy, it works with 1-minute data directly and handles internal resampling
# For other strategies, use their preferred timeframe
if primary_strategy.name == "bbrs":
# BBRS strategy processes 1-minute data and outputs signals on its internal timeframes
# Use 1-minute data for backtester working dataframe
working_df = data_1min.copy()
else:
# Other strategies specify their preferred timeframe
# Let the primary strategy resample the data to get the working dataframe
primary_strategy._resample_data(data_1min)
working_df = primary_strategy.get_primary_timeframe_data()
# Prepare working dataframe for backtester (ensure timestamp column)
working_df_for_backtest = working_df.copy().reset_index()
if 'index' in working_df_for_backtest.columns:
working_df_for_backtest = working_df_for_backtest.rename(columns={'index': 'timestamp'})
# Initialize backtest with strategy manager initialization
backtester = Backtest(initial_usd, working_df_for_backtest, working_df_for_backtest, strategy_manager_init)
# Store original min1_df for strategy processing
backtester.original_df = data_1min
# Attach strategy manager to backtester and initialize
backtester.strategy_manager = strategy_manager
strategy_manager.initialize(backtester)
n_trades = results["n_trades"] # Run backtest with strategy manager functions
trades = results.get('trades', []) results = backtester.run(
wins = [1 for t in trades if t['exit'] is not None and t['exit'] > t['entry']] strategy_manager_entry,
n_winning_trades = len(wins) strategy_manager_exit,
total_profit = sum(trade['profit_pct'] for trade in trades) debug
total_loss = sum(-trade['profit_pct'] for trade in trades if trade['profit_pct'] < 0) )
win_rate = n_winning_trades / n_trades if n_trades > 0 else 0
avg_trade = total_profit / n_trades if n_trades > 0 else 0
profit_ratio = total_profit / total_loss if total_loss > 0 else float('inf')
cumulative_profit = 0
max_drawdown = 0
peak = 0
for trade in trades: n_trades = results["n_trades"]
cumulative_profit += trade['profit_pct'] trades = results.get('trades', [])
wins = [1 for t in trades if t['exit'] is not None and t['exit'] > t['entry']]
n_winning_trades = len(wins)
total_profit = sum(trade['profit_pct'] for trade in trades)
total_loss = sum(-trade['profit_pct'] for trade in trades if trade['profit_pct'] < 0)
win_rate = n_winning_trades / n_trades if n_trades > 0 else 0
avg_trade = total_profit / n_trades if n_trades > 0 else 0
profit_ratio = total_profit / total_loss if total_loss > 0 else float('inf')
cumulative_profit = 0
max_drawdown = 0
peak = 0
if cumulative_profit > peak: for trade in trades:
peak = cumulative_profit cumulative_profit += trade['profit_pct']
drawdown = peak - cumulative_profit
if drawdown > max_drawdown: if cumulative_profit > peak:
max_drawdown = drawdown peak = cumulative_profit
drawdown = peak - cumulative_profit
final_usd = initial_usd if drawdown > max_drawdown:
max_drawdown = drawdown
for trade in trades: final_usd = initial_usd
final_usd *= (1 + trade['profit_pct'])
total_fees_usd = sum(trade.get('fee_usd', 0.0) for trade in trades) for trade in trades:
final_usd *= (1 + trade['profit_pct'])
# Update row to include timeframe information total_fees_usd = sum(trade.get('fee_usd', 0.0) for trade in trades)
row = {
"timeframe": f"{rule_name}({primary_timeframe})", # Show actual timeframe used # Get stop_loss_pct from the first strategy for reporting
# In multi-strategy setups, strategies can have different stop_loss_pct values
stop_loss_pct = primary_strategy.params.get("stop_loss_pct", "N/A")
# Update row to include timeframe information
row = {
"timeframe": f"{timeframe}({primary_timeframe})", # Show actual timeframe used
"stop_loss_pct": stop_loss_pct,
"n_trades": n_trades,
"n_stop_loss": sum(1 for trade in trades if 'type' in trade and trade['type'] == 'STOP_LOSS'),
"win_rate": win_rate,
"max_drawdown": max_drawdown,
"avg_trade": avg_trade,
"total_profit": total_profit,
"total_loss": total_loss,
"profit_ratio": profit_ratio,
"initial_usd": initial_usd,
"final_usd": final_usd,
"total_fees_usd": total_fees_usd,
}
results_rows.append(row)
for trade in trades:
trade_rows.append({
"timeframe": f"{timeframe}({primary_timeframe})",
"stop_loss_pct": stop_loss_pct, "stop_loss_pct": stop_loss_pct,
"n_trades": n_trades, "entry_time": trade.get("entry_time"),
"n_stop_loss": sum(1 for trade in trades if 'type' in trade and trade['type'] == 'STOP_LOSS'), "exit_time": trade.get("exit_time"),
"win_rate": win_rate, "entry_price": trade.get("entry"),
"max_drawdown": max_drawdown, "exit_price": trade.get("exit"),
"avg_trade": avg_trade, "profit_pct": trade.get("profit_pct"),
"total_profit": total_profit, "type": trade.get("type"),
"total_loss": total_loss, "fee_usd": trade.get("fee_usd"),
"profit_ratio": profit_ratio, })
"initial_usd": initial_usd,
"final_usd": final_usd, # Log strategy summary
"total_fees_usd": total_fees_usd, strategy_summary = strategy_manager.get_strategy_summary()
} logging.info(f"Timeframe: {timeframe}({primary_timeframe}), Stop Loss: {stop_loss_pct}, "
results_rows.append(row) f"Trades: {n_trades}, Strategies: {[s['name'] for s in strategy_summary['strategies']]}")
for trade in trades: if debug:
trade_rows.append({ # Plot after each backtest run
"timeframe": f"{rule_name}({primary_timeframe})", try:
"stop_loss_pct": stop_loss_pct, # Check if any strategy has processed_data for universal plotting
"entry_time": trade.get("entry_time"), processed_data = None
"exit_time": trade.get("exit_time"), for strategy in strategy_manager.strategies:
"entry_price": trade.get("entry"), if hasattr(backtester, 'processed_data') and backtester.processed_data is not None:
"exit_price": trade.get("exit"), processed_data = backtester.processed_data
"profit_pct": trade.get("profit_pct"), break
"type": trade.get("type"),
"fee_usd": trade.get("fee_usd"), if processed_data is not None and not processed_data.empty:
}) # Format strategy data with actual executed trades for universal plotting
formatted_data = BacktestCharts.format_strategy_data_with_trades(processed_data, results)
# Log strategy summary # Plot using universal function
strategy_summary = strategy_manager.get_strategy_summary() BacktestCharts.plot_data(formatted_data)
logging.info(f"Timeframe: {rule_name}({primary_timeframe}), Stop Loss: {stop_loss_pct}, " else:
f"Trades: {n_trades}, Strategies: {[s['name'] for s in strategy_summary['strategies']]}") # Fallback to meta_trend plot if available
if "meta_trend" in backtester.strategies:
if debug: meta_trend = backtester.strategies["meta_trend"]
# Plot after each backtest run # Use the working dataframe for plotting
try: BacktestCharts.plot(working_df, meta_trend)
# Check if any strategy has processed_data for universal plotting
processed_data = None
for strategy in strategy_manager.strategies:
if hasattr(backtester, 'processed_data') and backtester.processed_data is not None:
processed_data = backtester.processed_data
break
if processed_data is not None and not processed_data.empty:
# Format strategy data with actual executed trades for universal plotting
formatted_data = BacktestCharts.format_strategy_data_with_trades(processed_data, results)
# Plot using universal function
BacktestCharts.plot_data(formatted_data)
else: else:
# Fallback to meta_trend plot if available print("No plotting data available")
if "meta_trend" in backtester.strategies: except Exception as e:
meta_trend = backtester.strategies["meta_trend"] print(f"Plotting failed: {e}")
# Use the working dataframe for plotting
BacktestCharts.plot(working_df, meta_trend)
else:
print("No plotting data available")
except Exception as e:
print(f"Plotting failed: {e}")
return results_rows, trade_rows return results_rows, trade_rows
def process(timeframe_info, debug=False): def process(timeframe_info, debug=False):
"""Process a single (timeframe, stop_loss_pct) combination with strategy config""" """Process a single timeframe with strategy config"""
rule, data_1min, stop_loss_pct, initial_usd, strategy_config = timeframe_info timeframe, data_1min, config = timeframe_info
# Pass the original 1-minute data - strategies will handle their own timeframe resampling # Pass the essential data and full config
results_rows, all_trade_rows = process_timeframe_data( results_rows, all_trade_rows = process_timeframe_data(
data_1min, data_1min, [stop_loss_pct], rule, initial_usd, strategy_config, debug=debug data_1min, timeframe, config, debug=debug
) )
return results_rows, all_trade_rows return results_rows, all_trade_rows
@ -272,68 +258,25 @@ if __name__ == "__main__":
parser.add_argument("config", type=str, nargs="?", help="Path to config JSON file.") parser.add_argument("config", type=str, nargs="?", help="Path to config JSON file.")
args = parser.parse_args() args = parser.parse_args()
# Default values (from config.json) # Use config_default.json as fallback if no config provided
default_config = { config_file = args.config or "configs/config_default.json"
"start_date": "2025-03-01",
"stop_date": datetime.datetime.today().strftime('%Y-%m-%d'), try:
"initial_usd": 10000, with open(config_file, 'r') as f:
"timeframes": ["15min"],
"stop_loss_pcts": [0.03],
"strategies": [
{
"name": "default",
"weight": 1.0,
"params": {}
}
],
"combination_rules": {
"entry": "any",
"exit": "any",
"min_confidence": 0.5
}
}
if args.config:
with open(args.config, 'r') as f:
config = json.load(f) config = json.load(f)
elif not debug: print(f"Using config: {config_file}")
print("No config file provided. Please enter the following values (press Enter to use default):") except FileNotFoundError:
print(f"Error: Config file '{config_file}' not found.")
start_date = input(f"Start date [{default_config['start_date']}]: ") or default_config['start_date'] print("Available configs: configs/config_default.json, configs/config_bbrs.json, configs/config_combined.json")
stop_date = input(f"Stop date [{default_config['stop_date']}]: ") or default_config['stop_date'] exit(1)
except json.JSONDecodeError as e:
initial_usd_str = input(f"Initial USD [{default_config['initial_usd']}]: ") or str(default_config['initial_usd']) print(f"Error: Invalid JSON in config file '{config_file}': {e}")
initial_usd = float(initial_usd_str) exit(1)
timeframes_str = input(f"Timeframes (comma separated) [{', '.join(default_config['timeframes'])}]: ") or ','.join(default_config['timeframes'])
timeframes = [tf.strip() for tf in timeframes_str.split(',') if tf.strip()]
stop_loss_pcts_str = input(f"Stop loss pcts (comma separated) [{', '.join(str(x) for x in default_config['stop_loss_pcts'])}]: ") or ','.join(str(x) for x in default_config['stop_loss_pcts'])
stop_loss_pcts = [float(x.strip()) for x in stop_loss_pcts_str.split(',') if x.strip()]
config = {
'start_date': start_date,
'stop_date': stop_date,
'initial_usd': initial_usd,
'timeframes': timeframes,
'stop_loss_pcts': stop_loss_pcts,
'strategies': default_config['strategies'],
'combination_rules': default_config['combination_rules']
}
else:
config = default_config
start_date = config['start_date'] start_date = config['start_date']
stop_date = config['stop_date'] stop_date = config['stop_date']
initial_usd = config['initial_usd'] initial_usd = config['initial_usd']
timeframes = config['timeframes'] timeframes = config['timeframes']
stop_loss_pcts = config['stop_loss_pcts']
# Extract strategy configuration
strategy_config = {
"strategies": config.get('strategies', default_config['strategies']),
"combination_rules": config.get('combination_rules', default_config['combination_rules'])
}
timestamp = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M") timestamp = datetime.datetime.now().strftime("%Y_%m_%d_%H_%M")
@ -351,11 +294,10 @@ if __name__ == "__main__":
f"Initial USD\t{initial_usd}" f"Initial USD\t{initial_usd}"
] ]
# Create tasks for each (timeframe, stop_loss_pct) combination # Create tasks for each timeframe
tasks = [ tasks = [
(name, data_1min, stop_loss_pct, initial_usd, strategy_config) (name, data_1min, config)
for name in timeframes for name in timeframes
for stop_loss_pct in stop_loss_pcts
] ]
if debug: if debug: