# source: https://raw.githubusercontent.com/keithorange/FreqTradeCustomOrders/272d50e7d4c0eb7933caf035172c91ce54cb31d0/file_loading_strategy.py from datetime import datetime import json import time from typing import Any, Dict, Optional from freqtrade.persistence.trade_model import Order, Trade from freqtrade.strategy.interface import IStrategy from pandas import DataFrame import numpy as np import pandas_ta as pta from custom_order_form_handler import OrderStatus, StrategyDataHandler, ACTIVE_ORDER_STATUSES_VALUES from entry_conditions import price_crosses_upward, price_reverses_up, price_under from dateutil import parser import threading import logging class Github_keithorange_FreqTradeCustomOrders__file_loading_strategy__20240809_144149(IStrategy): """ Strategy class that reads order details from a file and sets strategy variables accordingly. """ # Default parameters stoploss = -1.0 def __init__(self, config) -> None: """ Initialize the strategy with the given configuration. """ super().__init__(config) self.strategy_name = self.__class__.__name__ self.order_handler = StrategyDataHandler( strategy_name=self.strategy_name) self.monitoring_initialized = False def input_strategy_data(self, pair: str): """ Placeholder method for handling argument input from file data handler. Implementation will vary based on specific strategy. """ raise NotImplementedError def set_entry_signal(self, pair: str, dataframe: DataFrame, data: Dict[str, Any]): """ Placeholder method for setting entry signal. Implementation will vary based on specific strategy. """ raise NotImplementedError def does_pair_have_data(self, pair) -> bool: data = self.order_handler.read_strategy_data() return pair in data def get_pair_data(self, pair) -> Dict[str, Any]: if not self.does_pair_have_data(pair): raise LookupError(f"{pair} doesn't have data yet!") return self.order_handler.read_strategy_data()[pair] def does_pair_have_active_order(self, pair) -> bool: if self.does_pair_have_data(pair): data = self.get_pair_data(pair) if data['status'] in ACTIVE_ORDER_STATUSES_VALUES: return True return False def get_dfile_arg(self, pair, key): data = self.get_pair_data(pair) if key in data: return data[key] return None def set_dfile_arg(self, pair, key, value): data = self.order_handler.read_strategy_data() if not self.does_pair_have_data(pair): data[pair] = {} data[pair][key] = value self.order_handler.save_strategy_data(data) def bot_loop_start(self, current_time: datetime, **kwargs) -> None: if not self.monitoring_initialized: self.start_monitoring() self.monitoring_initialized = True def start_monitoring(self): monitor_thread = threading.Thread(target=self.monitor_entry_conditions) monitor_thread.start() # Ensure last_prices is a list in monitor_entry_conditions def monitor_entry_conditions(self): while True: strategy_data = self.order_handler.read_strategy_data() for pair, data in strategy_data.items(): if data['status'] == OrderStatus.WAITING.value: condition_type = data['entry_condition'] entry_condition_timeout = data['entry_condition_timeout'] entry_condition_price = data.get('entry_condition_price') threshold_pct = data.get('threshold_pct', 0.15) ma_type = data.get('ma_type', 'EMA') period = data.get('period', 14) # USING EMA/HMA NOT CLOSE! last_prices = data.get('prices', []) # Check if prices is a LIST filled with price data! if not isinstance(last_prices, list) or len(last_prices) < 10 or not all(isinstance(price, (int, float)) for price in last_prices): logging.error(f"Invalid data type or insufficient data in last_prices: {type(last_prices)} for {pair}") continue # Check the condition if condition_type == 'PriceReversesUpCondition': is_satisfied = price_reverses_up( last_prices=last_prices, period=period, threshold_pct=threshold_pct, ) elif condition_type == 'PriceCrossesUpwardCondition': if not entry_condition_price: logging.error(f"Missing entry_condition_price for pair {pair}: {data}") continue is_satisfied = price_crosses_upward( entry_condition_price, last_prices=last_prices ) elif condition_type == 'PriceUnderCondition': if not entry_condition_price: logging.error(f"Missing entry_condition_price for pair {pair}: {data}") continue is_satisfied = price_under( entry_condition_price, last_prices=last_prices ) else: logging.error(f"Invalid condition type for pair {pair}: {data}") continue if is_satisfied: data['status'] = OrderStatus.PENDING.value if entry_condition_timeout and datetime.now() >= parser.parse(entry_condition_timeout): data['status'] = OrderStatus.CANCELED.value self.order_handler.update_strategy_data(pair, data) # SLEEP 2x per min! time.sleep(31) def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: pair = metadata['pair'] last_candle = dataframe.iloc[-1].squeeze() price = last_candle['close'] strategy_data = self.order_handler.read_strategy_data() # only enter if PENDING pair! if pair in strategy_data and strategy_data[pair]['status'] == OrderStatus.PENDING.value: # Enter trade freqtrade bot self.set_entry_signal(pair, dataframe, strategy_data[pair]) # write variabels to saved log strategy_data[pair]['status'] = OrderStatus.HOLDING.value strategy_data[pair]['entry_price'] = price # write to file self.order_handler.update_strategy_data(pair, strategy_data[pair]) else: self.set_no_entry(dataframe) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: return dataframe def set_no_entry(self, dataframe): dataframe.loc[dataframe.index[-1], ['enter_long', 'enter_tag']] = (0, "no_enter") def set_no_exit(self, dataframe): dataframe.loc[dataframe.index[-1], ['exit_long', 'exit_tag']] = (0, "no_exit") def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: default_stake = 10 # $10 if no stake is found try: return self.get_dfile_arg(pair, 'stake_amount') except ValueError as e: print(f"Error: {e}") return default_stake