86 lines
3.4 KiB
Python
86 lines
3.4 KiB
Python
import argparse
|
|
import logging
|
|
import sys
|
|
import time
|
|
import pandas as pd
|
|
import sqlite3
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
import importlib
|
|
|
|
from logging_utils import setup_logging
|
|
from strategies.base_strategy import BaseStrategy
|
|
|
|
class StrategyRunner:
|
|
"""
|
|
A generic runner that can execute any strategy that adheres to the
|
|
BaseStrategy blueprint. It handles the main logic loop, including data
|
|
loading, signal calculation, status saving, and sleeping.
|
|
"""
|
|
|
|
def __init__(self, strategy_name: str, log_level: str):
|
|
self.strategy_name = strategy_name
|
|
self.log_level = log_level
|
|
self.config = self._load_strategy_config()
|
|
if not self.config:
|
|
print(f"FATAL: Strategy '{strategy_name}' not found in configuration.")
|
|
sys.exit(1)
|
|
|
|
# Dynamically import and instantiate the strategy logic class
|
|
try:
|
|
module_path, class_name = self.config['class'].rsplit('.', 1)
|
|
module = importlib.import_module(module_path)
|
|
StrategyClass = getattr(module, class_name)
|
|
self.strategy_instance = StrategyClass(strategy_name, self.config['parameters'], self.log_level)
|
|
except (ImportError, AttributeError, KeyError) as e:
|
|
print(f"FATAL: Could not load strategy class for '{strategy_name}': {e}")
|
|
sys.exit(1)
|
|
|
|
def _load_strategy_config(self) -> dict:
|
|
"""Loads the configuration for the specified strategy."""
|
|
config_path = os.path.join("_data", "strategies.json")
|
|
try:
|
|
with open(config_path, 'r') as f:
|
|
all_configs = json.load(f)
|
|
return all_configs.get(self.strategy_name)
|
|
except (FileNotFoundError, json.JSONDecodeError) as e:
|
|
print(f"FATAL: Could not load strategy configuration: {e}")
|
|
return None
|
|
|
|
def run(self):
|
|
"""Main loop: loads data, calculates signals, saves status, and sleeps."""
|
|
logging.info(f"Starting main logic loop for {self.strategy_instance.coin} on {self.strategy_instance.timeframe}.")
|
|
while True:
|
|
df = self.strategy_instance.load_data()
|
|
if df.empty:
|
|
logging.warning("No data loaded. Waiting 1 minute before retrying...")
|
|
time.sleep(60)
|
|
continue
|
|
|
|
# The strategy instance calculates signals and updates its internal state
|
|
self.strategy_instance.calculate_signals_and_state(df.copy())
|
|
self.strategy_instance._save_status() # Save the new state
|
|
|
|
logging.info(f"Current Signal: {self.strategy_instance.current_signal}")
|
|
|
|
# Simple 1-minute wait for the next cycle
|
|
# A more precise timing mechanism could be implemented here if needed
|
|
time.sleep(60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="A generic runner for trading strategies.")
|
|
parser.add_argument("--name", required=True, help="The name of the strategy instance from strategies.json.")
|
|
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
runner = StrategyRunner(strategy_name=args.name, log_level=args.log_level)
|
|
runner.run()
|
|
except KeyboardInterrupt:
|
|
logging.info("Strategy runner stopped.")
|
|
except Exception as e:
|
|
logging.error(f"A critical error occurred in the strategy runner: {e}")
|
|
sys.exit(1)
|