progress

lib/pt_strategy/trading_strategy.py

@@ -309,96 +309,3 @@ class PtResearchStrategy:
 
     def day_trades(self) -> pd.DataFrame:
         return pd.concat(self.trades_, ignore_index=True)
-
-
-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=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)
-
-    # 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()
-
-
-    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()

research/backtest_new.py

@@ -0,0 +1,106 @@
+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
+
+
+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=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)
+
+    # 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()
+
+
+    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()