first stategies, import script for BTC histry

strategy_template.py

@@ -0,0 +1,132 @@
+import argparse
+import logging
+import sys
+import time
+import pandas as pd
+import sqlite3
+import json
+import os
+
+from logging_utils import setup_logging
+
+class TradingStrategy:
+    """
+    A template for a trading strategy that reads data from the SQLite database
+    and executes its logic in a loop.
+    """
+
+    def __init__(self, strategy_name: str, params: dict, log_level: str):
+        self.strategy_name = strategy_name
+        self.params = params
+        self.coin = params.get("coin", "N/A")
+        self.timeframe = params.get("timeframe", "N/A")
+        self.db_path = os.path.join("_data", "market_data.db")
+        
+        # Load strategy-specific parameters
+        self.rsi_period = params.get("rsi_period")
+        self.short_ma = params.get("short_ma")
+        self.long_ma = params.get("long_ma")
+        self.sma_period = params.get("sma_period")
+        
+        setup_logging(log_level, f"Strategy-{self.strategy_name}")
+        logging.info(f"Initializing strategy with parameters: {self.params}")
+
+    def load_data(self) -> pd.DataFrame:
+        """Loads historical data for the configured coin and timeframe from the database."""
+        table_name = f"{self.coin}_{self.timeframe}"
+        # Ensure we load enough data for the longest indicator period
+        limit = 500
+        if self.sma_period and self.sma_period > limit:
+            limit = self.sma_period + 50 # Add a buffer
+        elif self.long_ma and self.long_ma > limit:
+            limit = self.long_ma + 50
+
+        try:
+            with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn:
+                query = f'SELECT * FROM "{table_name}" ORDER BY datetime_utc DESC LIMIT {limit}'
+                df = pd.read_sql(query, conn)
+                df['datetime_utc'] = pd.to_datetime(df['datetime_utc'])
+                df.set_index('datetime_utc', inplace=True)
+                df.sort_index(inplace=True) # Ensure data is chronological
+                return df
+        except Exception as e:
+            logging.error(f"Failed to load data from table '{table_name}': {e}")
+            return pd.DataFrame()
+
+    def run_logic(self):
+        """
+        The main loop where the strategy's logic is executed.
+        This should be implemented with your specific trading rules.
+        """
+        logging.info(f"Starting main logic loop for {self.coin} on {self.timeframe} timeframe.")
+        while True:
+            data = self.load_data()
+
+            if data.empty:
+                logging.warning("No data loaded. Waiting before retrying...")
+                time.sleep(60)
+                continue
+
+            last_close = data['close'].iloc[-1]
+            logging.info(f"Latest data loaded. Last close price for {self.coin}: {last_close}")
+            
+            # --- SMA Strategy Logic ---
+            if self.sma_period:
+                if len(data) < self.sma_period:
+                    logging.warning(f"Not enough data to calculate {self.sma_period}-period SMA. "
+                                    f"Need {self.sma_period}, have {len(data)}.")
+                else:
+                    # Calculate the Simple Moving Average
+                    sma = data['close'].rolling(window=self.sma_period).mean().iloc[-1]
+                    logging.info(f"Current Price: {last_close}, {self.sma_period}-period SMA: {sma:.4f}")
+
+                    if last_close > sma:
+                        logging.warning("--- BUY SIGNAL --- (Price is above SMA)")
+                    elif last_close < sma:
+                        logging.warning("--- SELL SIGNAL --- (Price is below SMA)")
+                    else:
+                        logging.info("--- HOLD SIGNAL --- (Price is at SMA)")
+
+            # --- RSI Strategy Logic (Placeholder) ---
+            if self.rsi_period:
+                logging.info(f"RSI Period is set to: {self.rsi_period}. (RSI calculation not implemented).")
+
+            # --- MA Cross Strategy Logic (Placeholder) ---
+            if self.short_ma and self.long_ma:
+                logging.info(f"Short MA: {self.short_ma}, Long MA: {self.long_ma}. (MA Cross logic not implemented).")
+            
+            logging.info("Logic execution finished. Waiting for next cycle.")
+            time.sleep(60)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Run a trading strategy.")
+    parser.add_argument("--name", required=True, help="The name of the strategy instance from the config.")
+    parser.add_argument("--params", required=True, help="A JSON string of the strategy's parameters.")
+    parser.add_argument(
+        "--log-level",
+        default="normal",
+        choices=['off', 'normal', 'debug'],
+        help="Set the logging level for the script."
+    )
+    
+    args = parser.parse_args()
+
+    try:
+        strategy_params = json.loads(args.params)
+        strategy = TradingStrategy(
+            strategy_name=args.name, 
+            params=strategy_params,
+            log_level=args.log_level
+        )
+        strategy.run_logic()
+    except json.JSONDecodeError:
+        logging.error("Failed to decode JSON from --params argument.")
+        sys.exit(1)
+    except KeyboardInterrupt:
+        logging.info("Strategy process stopped.")
+        sys.exit(0)
+    except Exception as e:
+        logging.error(f"A critical error occurred: {e}")
+        sys.exit(1)
+