import argparse import hjson import importlib import glob import os import sqlite3 from datetime import datetime, date from typing import Any, Dict, List, Optional import pandas as pd from tools.data_loader import load_market_data from tools.trading_pair import TradingPair from results import BacktestResult, create_result_database, store_results_in_database def load_config(config_path: str) -> Dict: with open(config_path, "r") as f: config = hjson.load(f) return config def get_available_instruments_from_db(datafile: str, config: Dict) -> List[str]: """ Auto-detect available instruments from the database by querying distinct instrument_id values. Returns instruments without the configured prefix. """ try: conn = sqlite3.connect(datafile) # Query to get distinct instrument_ids query = f""" SELECT DISTINCT instrument_id FROM {config['db_table_name']} WHERE exchange_id = ? """ cursor = conn.execute(query, (config["exchange_id"],)) instrument_ids = [row[0] for row in cursor.fetchall()] conn.close() # Remove the configured prefix to get instrument symbols prefix = config.get("instrument_id_pfx", "") instruments = [] for instrument_id in instrument_ids: if instrument_id.startswith(prefix): symbol = instrument_id[len(prefix) :] instruments.append(symbol) else: instruments.append(instrument_id) return sorted(instruments) except Exception as e: print(f"Error auto-detecting instruments from {datafile}: {str(e)}") return [] def resolve_datafiles(config: Dict, cli_datafiles: Optional[str] = None) -> List[str]: """ Resolve the list of data files to process. CLI datafiles take priority over config datafiles. Supports wildcards in config but not in CLI. """ if cli_datafiles: # CLI override - comma-separated list, no wildcards datafiles = [f.strip() for f in cli_datafiles.split(",")] # Make paths absolute relative to data directory data_dir = config.get("data_directory", "./data") resolved_files = [] for df in datafiles: if not os.path.isabs(df): df = os.path.join(data_dir, df) resolved_files.append(df) return resolved_files # Use config datafiles with wildcard support config_datafiles = config.get("datafiles", []) data_dir = config.get("data_directory", "./data") resolved_files = [] for pattern in config_datafiles: if "*" in pattern or "?" in pattern: # Handle wildcards if not os.path.isabs(pattern): pattern = os.path.join(data_dir, pattern) matched_files = glob.glob(pattern) resolved_files.extend(matched_files) else: # Handle explicit file path if not os.path.isabs(pattern): pattern = os.path.join(data_dir, pattern) resolved_files.append(pattern) return sorted(list(set(resolved_files))) # Remove duplicates and sort def run_backtest( config: Dict, datafile: str, price_column: str, bt_result: BacktestResult, strategy, instruments: List[str], ) -> None: """ Run backtest for all pairs using the specified instruments. """ def _create_pairs(config: Dict, instruments: List[str]) -> List[TradingPair]: nonlocal datafile all_indexes = range(len(instruments)) unique_index_pairs = [(i, j) for i in all_indexes for j in all_indexes if i < j] pairs = [] # Update config to use the specified instruments config_copy = config.copy() config_copy["instruments"] = instruments market_data_df = load_market_data(datafile, config=config_copy) for a_index, b_index in unique_index_pairs: pair = TradingPair( market_data=market_data_df, symbol_a=instruments[a_index], symbol_b=instruments[b_index], price_column=price_column, ) pairs.append(pair) return pairs pairs_trades = [] for pair in _create_pairs(config, instruments): single_pair_trades = strategy.run_pair( pair=pair, config=config, bt_result=bt_result ) if single_pair_trades is not None and len(single_pair_trades) > 0: pairs_trades.append(single_pair_trades) # Check if result_list has any data before concatenating if len(pairs_trades) == 0: print("No trading signals found for any pairs") return None result = pd.concat(pairs_trades, ignore_index=True) result["time"] = pd.to_datetime(result["time"]) result = result.set_index("time").sort_index() bt_result.collect_single_day_results(result) def main() -> None: parser = argparse.ArgumentParser(description="Run pairs trading backtest.") parser.add_argument( "--config", type=str, required=True, help="Path to the configuration file." ) parser.add_argument( "--datafiles", type=str, required=False, help="Comma-separated list of data files (overrides config). No wildcards supported.", ) parser.add_argument( "--instruments", type=str, required=False, help="Comma-separated list of instrument symbols (e.g., COIN,GBTC). If not provided, auto-detects from database.", ) parser.add_argument( "--result_db", type=str, required=True, help="Path to SQLite database for storing results. Use 'NONE' to disable database output.", ) args = parser.parse_args() config: Dict = load_config(args.config) # Dynamically instantiate strategy class strategy_class_name = config.get("strategy_class", "strategies.StaticFitStrategy") module_name, class_name = strategy_class_name.rsplit(".", 1) module = importlib.import_module(module_name) strategy = getattr(module, class_name)() # Resolve data files (CLI takes priority over config) datafiles = resolve_datafiles(config, args.datafiles) if not datafiles: print("No data files found to process.") return print(f"Found {len(datafiles)} data files to process:") for df in datafiles: print(f" - {df}") # Create result database if needed if args.result_db.upper() != "NONE": create_result_database(args.result_db) # Initialize a dictionary to store all trade results all_results: Dict[str, Dict[str, Any]] = {} bt_results = BacktestResult(config=config) # Process each data file price_column = config["price_column"] for datafile in datafiles: print(f"\n====== Processing {os.path.basename(datafile)} ======") # Clear the trades for the new file bt_results.clear_trades() # Determine instruments to use if args.instruments: # Use CLI-specified instruments instruments = [inst.strip() for inst in args.instruments.split(",")] print(f"Using CLI-specified instruments: {instruments}") else: # Auto-detect instruments from database instruments = get_available_instruments_from_db(datafile, config) print(f"Auto-detected instruments: {instruments}") if not instruments: print(f"No instruments found for {datafile}, skipping...") continue # Process data for this file try: run_backtest( config=config, datafile=datafile, price_column=price_column, bt_result=bt_results, strategy=strategy, instruments=instruments, ) # Store results with file name as key filename = os.path.basename(datafile) all_results[filename] = {"trades": bt_results.trades.copy()} # Store results in database if args.result_db.upper() != "NONE": store_results_in_database(args.result_db, datafile, bt_results) print(f"Successfully processed {filename}") except Exception as e: print(f"Error processing {datafile}: {str(e)}") import traceback traceback.print_exc() # Calculate and print results if all_results: bt_results.calculate_returns(all_results) bt_results.print_grand_totals() bt_results.print_outstanding_positions() if args.result_db.upper() != "NONE": print(f"\nResults stored in database: {args.result_db}") else: print("No results to display.") if __name__ == "__main__": main()