ditus/hyper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

readme.md

Browse Source
This commit is contained in:
DiTus
2025-10-18 15:10:46 +02:00
parent 25df8b8ba9
commit 603a506c4e
11 changed files with 365 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
strategy_sma_cross.py
View File

@ -12,8 +12,9 @@ from logging_utils import setup_logging
class SmaCrossStrategy:
"""
A strategy that generates BUY/SELL signals based on the price crossing
a Simple Moving Average (SMA). It runs its logic precisely once per candle.
A flexible strategy that can operate in two modes:
1. Fast SMA / Slow SMA Crossover (if both 'fast' and 'slow' params are set)
2. Price / Single SMA Crossover (if only one 'fast' or 'slow' param is set)
"""
def __init__(self, strategy_name: str, params: dict, log_level: str):
@ -21,7 +22,10 @@ class SmaCrossStrategy:
self.params = params
self.coin = params.get("coin", "N/A")
self.timeframe = params.get("timeframe", "N/A")
self.sma_period = params.get("sma_period", 20) # Default to 20 if not specified
# Load fast and slow SMA periods, defaulting to 0 if not present
self.fast_ma_period = params.get("fast", 0)
self.slow_ma_period = params.get("slow", 0)
self.db_path = os.path.join("_data", "market_data.db")
self.status_file_path = os.path.join("_data", f"strategy_status_{self.strategy_name}.json")
@ -30,18 +34,25 @@ class SmaCrossStrategy:
self.current_signal = "INIT"
self.last_signal_change_utc = None
self.signal_price = None
self.indicator_value = None
self.fast_ma_value = None
self.slow_ma_value = None
setup_logging(log_level, f"Strategy-{self.strategy_name}")
logging.info(f"Initializing SMA Cross strategy with parameters:")
logging.info(f"Initializing SMA Crossover strategy with parameters:")
for key, value in self.params.items():
logging.info(f" - {key}: {value}")
def load_data(self) -> pd.DataFrame:
"""Loads historical data, ensuring enough for SMA calculation."""
"""Loads historical data, ensuring enough for the longest SMA calculation."""
table_name = f"{self.coin}_{self.timeframe}"
# We need at least sma_period + 1 rows to check the previous state
limit = self.sma_period + 50
# Determine the longest period needed for calculations
longest_period = max(self.fast_ma_period or 0, self.slow_ma_period or 0)
if longest_period == 0:
logging.error("No valid SMA periods ('fast' or 'slow' > 0) are defined in parameters.")
return pd.DataFrame()
limit = longest_period + 50
try:
with sqlite3.connect(f"file:{self.db_path}?mode=ro", uri=True) as conn:
@ -59,43 +70,57 @@ class SmaCrossStrategy:
def _calculate_signals(self, data: pd.DataFrame):
"""
Analyzes historical data to find the last SMA crossover event.
Analyzes historical data to find the last crossover event based on the
configured parameters (either dual or single SMA mode).
"""
if len(data) < self.sma_period + 1:
self.current_signal = "INSUFFICIENT DATA"
return
# --- DUAL SMA CROSSOVER LOGIC ---
if self.fast_ma_period and self.slow_ma_period:
if len(data) < self.slow_ma_period + 1:
self.current_signal = "INSUFFICIENT DATA"
return
# Calculate SMA
data['sma'] = data['close'].rolling(window=self.sma_period).mean()
self.indicator_value = data['sma'].iloc[-1]
data['fast_sma'] = data['close'].rolling(window=self.fast_ma_period).mean()
data['slow_sma'] = data['close'].rolling(window=self.slow_ma_period).mean()
self.fast_ma_value = data['fast_sma'].iloc[-1]
self.slow_ma_value = data['slow_sma'].iloc[-1]
# Position is 1 for Golden Cross (fast > slow), -1 for Death Cross
data['position'] = 0
data.loc[data['fast_sma'] > data['slow_sma'], 'position'] = 1
data.loc[data['fast_sma'] < data['slow_sma'], 'position'] = -1
# --- SINGLE SMA PRICE CROSS LOGIC ---
else:
sma_period = self.fast_ma_period or self.slow_ma_period
if len(data) < sma_period + 1:
self.current_signal = "INSUFFICIENT DATA"
return
data['sma'] = data['close'].rolling(window=sma_period).mean()
self.slow_ma_value = data['sma'].iloc[-1] # Use slow_ma_value to store the single SMA
self.fast_ma_value = None # Ensure fast is None
# Position is 1 when price is above SMA, -1 when below
data['position'] = 0
data.loc[data['close'] > data['sma'], 'position'] = 1
data.loc[data['close'] < data['sma'], 'position'] = -1
# Determine position relative to SMA: 1 for above (long), -1 for below (short)
data['position'] = 0
data.loc[data['close'] > data['sma'], 'position'] = 1
data.loc[data['close'] < data['sma'], 'position'] = -1
# A crossover is when the position on this candle is different from the last
# --- COMMON LOGIC for determining signal and last change ---
data['crossover'] = data['position'].diff()
# Get the latest signal based on the last position
last_position = data['position'].iloc[-1]
if last_position == 1: self.current_signal = "BUY"
elif last_position == -1: self.current_signal = "SELL"
else: self.current_signal = "HOLD"
# Find the most recent crossover event in the historical data
last_cross_series = data[data['crossover'] != 0]
if not last_cross_series.empty:
last_cross_row = last_cross_series.iloc[-1]
self.last_signal_change_utc = last_cross_row.name.tz_localize('UTC').isoformat()
self.signal_price = last_cross_row['close']
# Refine the signal to be the one *at the time of the cross*
if last_cross_row['position'] == 1: self.current_signal = "BUY"
elif last_cross_row['position'] == -1: self.current_signal = "SELL"
else:
# If no crosses in history, the signal has been consistent
self.last_signal_change_utc = data.index[0].tz_localize('UTC').isoformat()
self.signal_price = data['close'].iloc[0]
@ -122,15 +147,12 @@ class SmaCrossStrategy:
if tf_unit == 'm': interval_seconds = tf_value * 60
elif tf_unit == 'h': interval_seconds = tf_value * 3600
elif tf_unit == 'd': interval_seconds = tf_value * 86400
else: return 60 # Default to 1 minute if unknown
else: return 60
now = datetime.now(timezone.utc)
timestamp = now.timestamp()
# Calculate the timestamp of the *next* candle close
next_candle_ts = ((timestamp // interval_seconds) + 1) * interval_seconds
# Add a small buffer (e.g., 5 seconds) to ensure the candle data is available
sleep_seconds = (next_candle_ts - timestamp) + 5
logging.info(f"Next candle closes at {datetime.fromtimestamp(next_candle_ts, tz=timezone.utc)}. "
@ -139,7 +161,7 @@ class SmaCrossStrategy:
def run_logic(self):
"""Main loop: loads data, calculates signals, saves status, and sleeps."""
logging.info(f"Starting SMA Cross logic loop for {self.coin} on {self.timeframe} timeframe.")
logging.info(f"Starting logic loop for {self.coin} on {self.timeframe} timeframe.")
while True:
data = self.load_data()
if data.empty:
@ -152,14 +174,23 @@ class SmaCrossStrategy:
self._calculate_signals(data)
self._save_status()
# --- ADDED: More detailed logging for the current cycle ---
last_close = data['close'].iloc[-1]
indicator_val_str = f"{self.indicator_value:.4f}" if self.indicator_value is not None else "N/A"
logging.info(
f"Signal: {self.current_signal} | "
f"Price: {last_close:.4f} | "
f"SMA({self.sma_period}): {indicator_val_str}"
)
# --- Log based on which mode the strategy is running in ---
if self.fast_ma_period and self.slow_ma_period:
fast_ma_str = f"{self.fast_ma_value:.4f}" if self.fast_ma_value is not None else "N/A"
slow_ma_str = f"{self.slow_ma_value:.4f}" if self.slow_ma_value is not None else "N/A"
logging.info(
f"Signal: {self.current_signal} | Price: {last_close:.4f} | "
f"Fast SMA({self.fast_ma_period}): {fast_ma_str} | Slow SMA({self.slow_ma_period}): {slow_ma_str}"
)
else:
sma_period = self.fast_ma_period or self.slow_ma_period
sma_val_str = f"{self.slow_ma_value:.4f}" if self.slow_ma_value is not None else "N/A"
logging.info(
f"Signal: {self.current_signal} | Price: {last_close:.4f} | "
f"SMA({sma_period}): {sma_val_str}"
)
sleep_time = self.get_sleep_duration()
time.sleep(sleep_time)