# source: https://raw.githubusercontent.com/rmallarapu-bc/brahma/9287745fc036c2f4c00c586e598290885c2883b9/archive/Bhardwaja.py # --- Do not remove these libs --- from typing import Optional from pandas import DataFrame import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from datetime import datetime from freqtrade.persistence import Trade from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy from functools import reduce import logging import warnings import pandas as pd import requests log = logging.getLogger(__name__) log.setLevel(logging.INFO) warnings.simplefilter(action='ignore', category=pd.errors.PerformanceWarning) # -------------------------------- from datetime import datetime, time class Github_rmallarapu_bc_brahma__Bhardwaja__20240229_213751(IStrategy): INTERFACE_VERSION: int = 3 can_short = False # Exit criteria use_exit_signal = False exit_profit_only = False ignore_roi_if_entry_signal = True # ROI table: minimal_roi = {"0": 0.5} # Trailing stop: trailing_stop = True trailing_stop_positive = 0.02 trailing_stop_positive_offset = 0.1 trailing_only_offset_is_reached = True stoploss = -0.05 timeframe = '1d' # position_adjustment_enable = True # max_entry_position_adjustment = 10 fear_index_long = IntParameter(1, 6, default=4, space="buy") fear_index_short = IntParameter(1, 6, default=1, space="sell") feardf = pd.DataFrame() def fear_index(self, dataframe, feardf): prev_resp = { "name": "Fear and Greed Index", "data": [ { "value": "3", "value_classification": "Neutral", "timestamp": str(datetime.today()), } ] } if feardf.empty: resp = requests.get('https://api.alternative.me/fng/?limit=9999&date_format=kr') else: if datetime.today() in feardf['date'].values: return feardf['value_classification'] else: resp = requests.get('https://api.alternative.me/fng/?limit=1&date_format=kr') if resp is not None and resp.headers.get('Content-Type').startswith('application/json'): try: self.prev_resp = resp.json() df_gf = self.prev_resp['data'] except: self.prev_resp = prev_resp df_gf = self.prev_resp['data'] else: self.prev_resp = prev_resp df_gf = self.prev_resp['data'] fear = pd.json_normalize(df_gf) fear.rename(columns={'value': 'fear'}, inplace=True) fear.rename(columns={'timestamp': 'date'}, inplace=True) fear['value_classification'] = fear['value_classification'].replace('Extreme Fear', 5) fear['value_classification'] = fear['value_classification'].replace('Fear', 4) fear['value_classification'] = fear['value_classification'].replace('Neutral', 3) fear['value_classification'] = fear['value_classification'].replace('Greed', 2) fear['value_classification'] = fear['value_classification'].replace('Extreme Greed', 1) df_copy = dataframe[['date']].copy() df_copy["date"] = pd.to_datetime(df_copy["date"], unit='ms') df_copy['date'] = df_copy['date'].astype(str) df_copy['date'] = df_copy['date'].str[:10] feardf = pd.merge(df_copy, fear, on='date') return feardf['value_classification'] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['fear_index'] = self.fear_index(dataframe, self.feardf) log.info(dataframe['fear_index'].nunique()) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: log.info(dataframe['fear_index'].nunique()) dataframe.loc[ ( (dataframe['fear_index'] == self.fear_index_long.value) # & # (dataframe['fear_index'] > 1) # (dataframe['fear_index'].between(0, 4, inclusive='both').any()) # & # (dataframe['date'].dt.hour == 0) ), 'enter_long'] = 1 dataframe.loc[ ( (dataframe['fear_index'] == self.fear_index_short.value) # & # (dataframe['fear_index'] <= 5) # (dataframe['fear_index'].between(0, 4, inclusive='both').any()) # & # (dataframe['date'].dt.hour == 0) ), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( # (dataframe['fear_index'].between(3, 6, inclusive='both').any()) # (dataframe['fear_index'].between(4, 5).any()) (dataframe['fear_index'] == self.fear_index_short.value) # & # (dataframe['fear_index'] <= 5) # & # (dataframe['date'].dt.hour == 23) ), 'exit_long'] = 1 dataframe.loc[ ( # (dataframe['fear_index'].between(3, 6, inclusive='both').any()) # (dataframe['fear_index'].between(4, 5).any()) (dataframe['fear_index'] == self.fear_index_long.value) # & # (dataframe['fear_index'] > 1) # & # (dataframe['date'].dt.hour == 23) ), 'exit_short'] = 1 return dataframe initial_order_size = 1.00 position_adjustment_enable = False max_entry_position_adjustment = 2 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: return proposed_stake * self.initial_order_size def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs) -> Optional[float]: """ Custom trade adjustment logic, returning the stake amount that a trade should be increased or decreased. This means extra entry or exit orders with additional fees. Only called when `position_adjustment_enable` is set to True. For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/ When not implemented by a strategy, returns None :param trade: trade object. :param current_time: datetime object, containing the current datetime :param current_rate: Current entry rate (same as current_entry_profit) :param current_profit: Current profit (as ratio), calculated based on current_rate (same as current_entry_profit). :param min_stake: Minimal stake size allowed by exchange (for both entries and exits) :param max_stake: Maximum stake allowed (either through balance, or by exchange limits). :param current_entry_rate: Current rate using entry pricing. :param current_exit_rate: Current rate using exit pricing. :param current_entry_profit: Current profit using entry pricing. :param current_exit_profit: Current profit using exit pricing. :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. :return float: Stake amount to adjust your trade, Positive values to increase position, Negative values to decrease position. Return None for no action. """ count_of_entries = trade.nr_of_successful_entries filled_entries = trade.select_filled_orders(trade.entry_side) if current_profit > 0.1 and count_of_entries <= self.max_entry_position_adjustment: try: stake_amount = filled_entries[0].stake_amount stake_amount = stake_amount * 0.01 # log.info(f"New stake_amount = {stake_amount}") return stake_amount except Exception as exception: return None dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) # Only buy when not actively falling price. last_candle = dataframe.iloc[-1].squeeze() previous_candle = dataframe.iloc[-2].squeeze() if last_candle['close'] < previous_candle['close']: return None # filled_entries = trade.select_filled_orders(trade.entry_side) if current_profit < -0.1 and count_of_entries <= self.max_entry_position_adjustment: # log.info(f"total_profit = {trade.total_profit}") # log.info(f"Current profit = {current_profit}") try: stake_amount = filled_entries[0].stake_amount stake_amount = stake_amount * 0.01 # log.info(f"New stake_amount = {stake_amount}") return stake_amount except Exception as exception: return None return None