112 lines
3.3 KiB
Python
112 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any, Dict
|
|
|
|
from pt_strategy.results import (PairResearchResult, create_result_database,
|
|
store_config_in_database)
|
|
from pt_strategy.trading_strategy import PtResearchStrategy
|
|
from tools.filetools import resolve_datafiles
|
|
from tools.instruments import get_instruments
|
|
from tools.viz.viz_trades import visualize_trades
|
|
|
|
|
|
def main() -> None:
|
|
import argparse
|
|
|
|
from tools.config import expand_filename, load_config
|
|
|
|
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(
|
|
"--date_pattern",
|
|
type=str,
|
|
required=True,
|
|
help="Date YYYYMMDD, allows * and ? wildcards",
|
|
)
|
|
parser.add_argument(
|
|
"--instruments",
|
|
type=str,
|
|
required=True,
|
|
help="Comma-separated list of instrument symbols (e.g., COIN:EQUITY,GBTC:CRYPTO)",
|
|
)
|
|
parser.add_argument(
|
|
"--result_db",
|
|
type=str,
|
|
required=False,
|
|
default="NONE",
|
|
help="Path to SQLite database for storing results. Use 'NONE' to disable database output.",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
config: Dict = load_config(args.config)
|
|
|
|
# Resolve data files (CLI takes priority over config)
|
|
instruments = get_instruments(args, config)
|
|
datafiles = resolve_datafiles(config, args.date_pattern, instruments)
|
|
|
|
days = list(set([day for day, _ in datafiles]))
|
|
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":
|
|
args.result_db = expand_filename(args.result_db)
|
|
create_result_database(args.result_db)
|
|
|
|
# Initialize a dictionary to store all trade results
|
|
all_results: Dict[str, Dict[str, Any]] = {}
|
|
is_config_stored = False
|
|
# Process each data file
|
|
|
|
results = PairResearchResult(config=config)
|
|
for day in sorted(days):
|
|
md_datafiles = [datafile for md_day, datafile in datafiles if md_day == day]
|
|
if not all([os.path.exists(datafile) for datafile in md_datafiles]):
|
|
print(f"WARNING: insufficient data files: {md_datafiles}")
|
|
continue
|
|
print(f"\n====== Processing {day} ======")
|
|
|
|
if not is_config_stored:
|
|
store_config_in_database(
|
|
db_path=args.result_db,
|
|
config_file_path=args.config,
|
|
config=config,
|
|
datafiles=datafiles,
|
|
instruments=instruments,
|
|
)
|
|
is_config_stored = True
|
|
|
|
pt_strategy = PtResearchStrategy(
|
|
config=config, datafiles=md_datafiles, instruments=instruments
|
|
)
|
|
pt_strategy.run()
|
|
results.add_day_results(
|
|
day=day,
|
|
trades=pt_strategy.day_trades(),
|
|
outstanding_positions=pt_strategy.outstanding_positions(),
|
|
)
|
|
|
|
|
|
results.analyze_pair_performance()
|
|
|
|
|
|
visualize_trades(pt_strategy, results, day)
|
|
|
|
|
|
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()
|