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

strategy status table

Browse Source
This commit is contained in:
DiTus
2025-10-15 18:32:12 +02:00
parent bbfb549fbb
commit 0d53200882
15 changed files with 464 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
_data/market_cap_data.json
View File

@ -36,12 +36,12 @@
"market_cap": 10637373991.458858
},
"TOTAL_market_cap_daily": {
"datetime_utc": "2025-10-14 00:00:00",
"market_cap": 3942937396387.7046
"datetime_utc": "2025-10-15 00:00:00",
"market_cap": 3950478733651.1655
},
"PUMP_market_cap": {
"datetime_utc": "2025-10-14 21:02:30",
"market_cap": 1454398647.593871
},
"summary_last_updated_utc": "2025-10-14T21:08:01.788055+00:00"
"summary_last_updated_utc": "2025-10-15T00:16:07.128221+00:00"
}

BIN
_data/market_data.db-shm
View File

Binary file not shown.

29
_data/strategies.json
View File

@ -8,14 +8,24 @@
"sma_period": 125
}
},
"ma_cross_btc": {
"sma_cross_1": {
"enabled": true,
"script": "strategy_template.py",
"script": "strategy_sma_cross.py",
"parameters": {
"coin": "ETH",
"timeframe": "5m",
"sma_period": 5,
"rma_period": 10,
"ema_period": 15
}
},
"sma_cross_2": {
"enabled": true,
"script": "strategy_sma_cross.py",
"parameters": {
"coin": "BTC",
"timeframe": "1h",
"short_ma": 10,
"long_ma": 50
"timeframe": "5m",
"sma_period": 5
}
},
"sma_125d_btc": {
@ -36,12 +46,13 @@
"sma_period": 44
}
},
"disabled_strategy": {
"enabled": false,
"sma_5m_eth": {
"enabled": true,
"script": "strategy_template.py",
"parameters": {
"coin": "SOL",
"timeframe": "15m"
"coin": "ETH",
"timeframe": "5m",
"sma_period": 5
}
}
}

7
_data/strategy_status_ma_cross_btc.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "ma_cross_btc",
"current_signal": "HOLD",
"last_signal_change_utc": "2025-10-12T17:00:00+00:00",
"signal_price": 114286.0,
"last_checked_utc": "2025-10-15T11:48:55.092260+00:00"
}

7
_data/strategy_status_sma_125d_btc.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_125d_btc",
"current_signal": "SELL",
"last_signal_change_utc": "2025-10-14T00:00:00+00:00",
"signal_price": 113026.0,
"last_checked_utc": "2025-10-15T16:31:15.415923+00:00"
}

7
_data/strategy_status_sma_125d_eth.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_125d_eth",
"current_signal": "BUY",
"last_signal_change_utc": "2025-08-26T00:00:00+00:00",
"signal_price": 4600.63,
"last_checked_utc": "2025-10-15T16:31:15.411175+00:00"
}

7
_data/strategy_status_sma_44d_btc.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_44d_btc",
"current_signal": "SELL",
"last_signal_change_utc": "2025-10-14T00:00:00+00:00",
"signal_price": 113026.0,
"last_checked_utc": "2025-10-15T16:31:15.422945+00:00"
}

7
_data/strategy_status_sma_5m_eth.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_5m_eth",
"current_signal": "BUY",
"last_signal_change_utc": "2025-10-15T16:00:00+00:00",
"signal_price": 3976.4,
"last_checked_utc": "2025-10-15T16:30:15.367655+00:00"
}

7
_data/strategy_status_sma_cross.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_cross",
"current_signal": "SELL",
"last_signal_change_utc": "2025-10-15T11:45:00+00:00",
"signal_price": 111957.0,
"last_checked_utc": "2025-10-15T12:10:05.048434+00:00"
}

7
_data/strategy_status_sma_cross_1.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_cross_1",
"current_signal": "BUY",
"last_signal_change_utc": "2025-10-15T16:00:00+00:00",
"signal_price": 3976.4,
"last_checked_utc": "2025-10-15T16:30:15.368224+00:00"
}

7
_data/strategy_status_sma_cross_2.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_cross_2",
"current_signal": "BUY",
"last_signal_change_utc": "2025-10-15T16:25:00+00:00",
"signal_price": 111016.0,
"last_checked_utc": "2025-10-15T16:30:15.380563+00:00"
}

7
_data/strategy_status_sma_cross_eth_5m.json Normal file
View File

@ -0,0 +1,7 @@
{
"strategy_name": "sma_cross_eth_5m",
"current_signal": "SELL",
"last_signal_change_utc": "2025-10-15T11:45:00+00:00",
"signal_price": 4106.1,
"last_checked_utc": "2025-10-15T12:05:05.022308+00:00"
}

109
main_app.py
View File

@ -133,7 +133,7 @@ def run_strategy(strategy_name: str, config: dict):
log_file = os.path.join(LOGS_DIR, f"strategy_{strategy_name}.log")
script_name = config['script']
params_str = json.dumps(config['parameters'])
command = [sys.executable, script_name, "--name", strategy_name, "--params", params_str, "--log-level", "off"]
command = [sys.executable, script_name, "--name", strategy_name, "--params", params_str, "--log-level", "normal"]
while True:
try:
with open(log_file, 'a') as f:
@ -147,14 +147,15 @@ def run_strategy(strategy_name: str, config: dict):
class MainApp:
def __init__(self, coins_to_watch: list, processes: dict):
def __init__(self, coins_to_watch: list, processes: dict, strategy_configs: dict):
self.watched_coins = coins_to_watch
self.prices = {}
self.market_caps = {}
self.last_db_update_info = "Initializing..."
self._lines_printed = 0
self.background_processes = processes
self.process_status = {}
self.strategy_configs = strategy_configs
self.strategy_statuses = {}
def read_prices(self):
"""Reads the latest prices from the JSON file."""
@ -172,7 +173,6 @@ class MainApp:
with open(MARKET_CAP_SUMMARY_FILE, 'r', encoding='utf-8') as f:
summary_data = json.load(f)
# Extract just the market cap value for each coin
for coin in self.watched_coins:
table_key = f"{coin}_market_cap"
if table_key in summary_data:
@ -180,6 +180,20 @@ class MainApp:
except (json.JSONDecodeError, IOError):
logging.debug("Could not read market cap summary file.")
def read_strategy_statuses(self):
"""Reads the status JSON file for each enabled strategy."""
for name in self.strategy_configs.keys():
status_file = os.path.join("_data", f"strategy_status_{name}.json")
if os.path.exists(status_file):
try:
with open(status_file, 'r', encoding='utf-8') as f:
self.strategy_statuses[name] = json.load(f)
except (IOError, json.JSONDecodeError):
self.strategy_statuses[name] = {"error": "Could not read status file."}
else:
self.strategy_statuses[name] = {"current_signal": "Initializing..."}
def get_overall_db_status(self):
"""Reads the fetcher status from the status file."""
if os.path.exists(STATUS_FILE):
@ -210,37 +224,70 @@ class MainApp:
self.process_status[name] = "Running" if process.is_alive() else "STOPPED"
def display_dashboard(self):
"""Displays a formatted table without blinking by overwriting previous lines."""
if self._lines_printed > 0:
print(f"\x1b[{self._lines_printed}A", end="")
output_lines = ["--- Market Dashboard ---"]
table_width = 44
output_lines.append("-" * table_width)
output_lines.append(f"{'#':<2} | {'Coin':<6} | {'Live Price':>10} | {'Market Cap':>15} |")
output_lines.append("-" * table_width)
"""Displays a formatted dashboard with side-by-side tables."""
print("\x1b[H\x1b[J", end="") # Clear screen
# --- Build Left Table (Market Dashboard) ---
left_table_lines = []
left_table_width = 44
left_table_lines.append("--- Market Dashboard ---\t\t")
left_table_lines.append("-" * left_table_width)
left_table_lines.append(f"{'#':^2} | {'Coin':^6} | {'Live Price':>10} | {'Market Cap':>15} |")
left_table_lines.append("-" * left_table_width)
for i, coin in enumerate(self.watched_coins, 1):
price = self.prices.get(coin, "Loading...")
market_cap = self.market_caps.get(coin)
formatted_mc = format_market_cap(market_cap)
output_lines.append(f"{i:<2} | {coin:<6} | {price:>10} | {formatted_mc:>15} |")
output_lines.append("-" * table_width)
left_table_lines.append(f"{i:<2} | {coin:^6} | {price:>10} | {formatted_mc:>15} |")
left_table_lines.append("-" * left_table_width)
# --- Build Right Table (Strategy Status) ---
right_table_lines = []
right_table_width = 148
right_table_lines.append("--- Strategy Status ---")
right_table_lines.append("-" * right_table_width)
right_table_lines.append(f"{'#':<2} | {'Strategy Name':<25} | {'Coin':^6} | {'Signal':<8} | {'Signal Price':>12} | {'Last Change (Local)':>22} | {'TF':^5} | {'Parameters':<45} |")
right_table_lines.append("-" * right_table_width)
for i, (name, status) in enumerate(self.strategy_statuses.items(), 1):
signal = status.get('current_signal', 'N/A')
price = status.get('signal_price')
price_display = f"{price:.4f}" if isinstance(price, (int, float)) else "-"
last_change = status.get('last_signal_change_utc')
last_change_display = 'Never'
if last_change:
# Convert UTC timestamp from file to local time for display
dt_utc = datetime.fromisoformat(last_change.replace('Z', '+00:00')).replace(tzinfo=timezone.utc)
dt_local = dt_utc.astimezone(None)
last_change_display = dt_local.strftime('%Y-%m-%d %H:%M')
config_params = self.strategy_configs.get(name, {}).get('parameters', {})
coin = config_params.get('coin', 'N/A')
timeframe = config_params.get('timeframe', 'N/A')
other_params = {k: v for k, v in config_params.items() if k not in ['coin', 'timeframe']}
params_str = ", ".join([f"{k}={v}" for k, v in other_params.items()])
right_table_lines.append(f"{i:^2} | {name:<25} | {coin:^6} | {signal:<8} | {price_display:>12} | {last_change_display:>22} | {timeframe:^5} | {params_str:<45} |")
right_table_lines.append("-" * right_table_width)
status_prefix = "DB Status: Last update -> "
max_len = 80
status_message = f"{status_prefix}{self.last_db_update_info}"
if len(status_message) > max_len:
status_message = status_message[:max_len-3] + "..."
output_lines.append(status_message)
output_lines.append("--- Background Processes ---")
# --- Combine Tables Side-by-Side ---
output_lines = []
max_rows = max(len(left_table_lines), len(right_table_lines))
separator = " "
indent = " " * 10
for i in range(max_rows):
left_part = left_table_lines[i] if i < len(left_table_lines) else " " * left_table_width
right_part = indent + right_table_lines[i] if i < len(right_table_lines) else ""
output_lines.append(f"{left_part}{separator}{right_part}")
# --- Add Bottom Sections ---
output_lines.append(f"\nDB Status: Last update -> {self.last_db_update_info}")
output_lines.append("\n--- Background Processes ---")
for name, status in self.process_status.items():
output_lines.append(f"{name:<25}: {status}")
final_output = "\n".join(output_lines) + "\n\x1b[J"
print(final_output, end="")
self._lines_printed = len(output_lines)
final_output = "\n".join(output_lines)
print(final_output)
sys.stdout.flush()
def run(self):
@ -249,6 +296,7 @@ class MainApp:
self.read_prices()
self.read_market_caps()
self.get_overall_db_status()
self.read_strategy_statuses()
self.check_process_status()
self.display_dashboard()
time.sleep(2)
@ -268,6 +316,7 @@ if __name__ == "__main__":
sys.exit(1)
processes = {}
strategy_configs = {}
processes["Market Feeder"] = multiprocessing.Process(target=run_market_feeder, daemon=True)
processes["Data Fetcher"] = multiprocessing.Process(target=data_fetcher_scheduler, daemon=True)
@ -279,18 +328,22 @@ if __name__ == "__main__":
strategy_configs = json.load(f)
for name, config in strategy_configs.items():
if config.get("enabled", False):
if not os.path.exists(config['script']):
logging.error(f"Strategy script '{config['script']}' for strategy '{name}' not found. Skipping.")
continue
proc = multiprocessing.Process(target=run_strategy, args=(name, config), daemon=True)
processes[f"Strategy: {name}"] = proc
except (FileNotFoundError, json.JSONDecodeError) as e:
logging.error(f"Could not load strategies from '{STRATEGY_CONFIG_FILE}': {e}")
# Launch all processes
for name, proc in processes.items():
logging.info(f"Starting process '{name}'...")
proc.start()
time.sleep(3)
app = MainApp(coins_to_watch=WATCHED_COINS, processes=processes)
app = MainApp(coins_to_watch=WATCHED_COINS, processes=processes, strategy_configs=strategy_configs)
try:
app.run()
except KeyboardInterrupt:

188
strategy_sma_cross.py Normal file
View File

@ -0,0 +1,188 @@
import argparse
import logging
import sys
import time
import pandas as pd
import sqlite3
import json
import os
from datetime import datetime, timezone, timedelta
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.
"""
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.sma_period = params.get("sma_period", 20) # Default to 20 if not specified
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")
# Strategy state variables
self.current_signal = "INIT"
self.last_signal_change_utc = None
self.signal_price = None
self.indicator_value = None
setup_logging(log_level, f"Strategy-{self.strategy_name}")
logging.info(f"Initializing SMA Cross 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."""
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
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)
if df.empty: return pd.DataFrame()
df['datetime_utc'] = pd.to_datetime(df['datetime_utc'])
df.set_index('datetime_utc', inplace=True)
df.sort_index(inplace=True)
return df
except Exception as e:
logging.error(f"Failed to load data from table '{table_name}': {e}")
return pd.DataFrame()
def _calculate_signals(self, data: pd.DataFrame):
"""
Analyzes historical data to find the last SMA crossover event.
"""
if len(data) < self.sma_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]
# 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
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]
def _save_status(self):
"""Saves the current strategy state to its JSON file."""
status = {
"strategy_name": self.strategy_name,
"current_signal": self.current_signal,
"last_signal_change_utc": self.last_signal_change_utc,
"signal_price": self.signal_price,
"last_checked_utc": datetime.now(timezone.utc).isoformat()
}
try:
with open(self.status_file_path, 'w', encoding='utf-8') as f:
json.dump(status, f, indent=4)
except IOError as e:
logging.error(f"Failed to write status file: {e}")
def get_sleep_duration(self) -> int:
"""Calculates seconds to sleep until the next full candle closes."""
tf_value = int(''.join(filter(str.isdigit, self.timeframe)))
tf_unit = ''.join(filter(str.isalpha, self.timeframe))
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
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)}. "
f"Sleeping for {sleep_seconds:.2f} seconds.")
return sleep_seconds
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.")
while True:
data = self.load_data()
if data.empty:
logging.warning("No data loaded. Waiting 1 minute before retrying...")
self.current_signal = "NO DATA"
self._save_status()
time.sleep(60)
continue
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}"
)
sleep_time = self.get_sleep_duration()
time.sleep(sleep_time)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Run an SMA Crossover 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'])
args = parser.parse_args()
try:
strategy_params = json.loads(args.params)
strategy = SmaCrossStrategy(
strategy_name=args.name,
params=strategy_params,
log_level=args.log_level
)
strategy.run_logic()
except KeyboardInterrupt:
logging.info("Strategy process stopped.")
except Exception as e:
logging.error(f"A critical error occurred: {e}")
sys.exit(1)

164
strategy_template.py
View File

@ -6,13 +6,14 @@ import pandas as pd
import sqlite3
import json
import os
from datetime import datetime, timezone, timedelta
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.
and executes its logic in a loop, running once per candle.
"""
def __init__(self, strategy_name: str, params: dict, log_level: str):
@ -21,8 +22,15 @@ class TradingStrategy:
self.coin = params.get("coin", "N/A")
self.timeframe = params.get("timeframe", "N/A")
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")
# Load strategy-specific parameters
# Strategy state variables
self.current_signal = "INIT"
self.last_signal_change_utc = None
self.signal_price = None
self.indicator_value = None
# Load strategy-specific parameters from config
self.rsi_period = params.get("rsi_period")
self.short_ma = params.get("short_ma")
self.long_ma = params.get("long_ma")
@ -32,84 +40,134 @@ class TradingStrategy:
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."""
"""Loads historical data, ensuring enough for the longest indicator period."""
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
# Determine required data limit based on the longest configured indicator
periods = [p for p in [self.sma_period, self.long_ma, self.rsi_period] if p is not None]
if periods:
limit = max(periods) + 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)
if df.empty: return pd.DataFrame()
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
df.sort_index(inplace=True)
return df
except Exception as e:
logging.error(f"Failed to load data from table '{table_name}': {e}")
return pd.DataFrame()
def _calculate_signals(self, data: pd.DataFrame):
"""
Analyzes historical data to find the last signal crossover event.
This method should be expanded to handle different strategy types.
"""
if self.sma_period:
if len(data) < self.sma_period + 1:
self.current_signal = "INSUFFICIENT DATA"
return
data['sma'] = data['close'].rolling(window=self.sma_period).mean()
self.indicator_value = data['sma'].iloc[-1]
data['position'] = 0
data.loc[data['close'] > data['sma'], 'position'] = 1
data.loc[data['close'] < data['sma'], 'position'] = -1
data['crossover'] = data['position'].diff()
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"
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']
if last_cross_row['position'] == 1: self.current_signal = "BUY"
elif last_cross_row['position'] == -1: self.current_signal = "SELL"
else:
self.last_signal_change_utc = data.index[0].tz_localize('UTC').isoformat()
self.signal_price = data['close'].iloc[0]
elif self.rsi_period:
logging.info(f"RSI logic not implemented for period {self.rsi_period}.")
self.current_signal = "NOT IMPLEMENTED"
elif self.short_ma and self.long_ma:
logging.info(f"MA Cross logic not implemented for {self.short_ma}/{self.long_ma}.")
self.current_signal = "NOT IMPLEMENTED"
def _save_status(self):
"""Saves the current strategy state to its JSON file."""
status = {
"strategy_name": self.strategy_name,
"current_signal": self.current_signal,
"last_signal_change_utc": self.last_signal_change_utc,
"signal_price": self.signal_price,
"last_checked_utc": datetime.now(timezone.utc).isoformat()
}
try:
with open(self.status_file_path, 'w', encoding='utf-8') as f:
json.dump(status, f, indent=4)
except IOError as e:
logging.error(f"Failed to write status file: {e}")
def get_sleep_duration(self) -> int:
"""Calculates seconds to sleep until the next full candle closes."""
if not self.timeframe: return 60
tf_value = int(''.join(filter(str.isdigit, self.timeframe)))
tf_unit = ''.join(filter(str.isalpha, self.timeframe))
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
now = datetime.now(timezone.utc)
timestamp = now.timestamp()
next_candle_ts = ((timestamp // interval_seconds) + 1) * interval_seconds
sleep_seconds = (next_candle_ts - timestamp) + 5
logging.info(f"Next candle closes at {datetime.fromtimestamp(next_candle_ts, tz=timezone.utc)}. "
f"Sleeping for {sleep_seconds:.2f} seconds.")
return sleep_seconds
def run_logic(self):
"""
The main loop where the strategy's logic is executed.
This should be implemented with your specific trading rules.
"""
"""Main loop: loads data, calculates signals, saves status, and sleeps."""
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...")
logging.warning("No data loaded. Waiting 1 minute before retrying...")
self.current_signal = "NO DATA"
self._save_status()
time.sleep(60)
continue
self._calculate_signals(data)
self._save_status()
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}")
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} | Price: {last_close:.4f} | Indicator: {indicator_val_str}")
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)
sleep_time = self.get_sleep_duration()
time.sleep(sleep_time)
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."
)
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
args = parser.parse_args()
try:
@ -120,12 +178,8 @@ if __name__ == "__main__":
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)