# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these imports --- import numpy as np import pandas as pd from datetime import datetime, timedelta, timezone from pandas import DataFrame from typing import Dict, Optional, Union, Tuple from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, # @informative decorator # Hyperopt Parameters BooleanParameter, CategoricalParameter, DecimalParameter, IntParameter, RealParameter, # timeframe helpers timeframe_to_minutes, timeframe_to_next_date, timeframe_to_prev_date, # Strategy helper functions merge_informative_pair, stoploss_from_absolute, stoploss_from_open, ) # -------------------------------- # Add your lib to import here import talib.abstract as ta import pandas_ta as pta from technical import qtpylib class PriceVolume1MNoticeStrategy(IStrategy): """ This is a strategy template to get you started. More information in https://www.freqtrade.io/en/latest/strategy-customization/ You can: :return: a Dataframe with all mandatory indicators for the strategies - Rename the class name (Do not forget to update class_name) - Add any methods you want to build your strategy - Add any lib you need to build your strategy You must keep: - the lib in the section "Do not remove these libs" - the methods: populate_indicators, populate_entry_trend, populate_exit_trend You should keep: - timeframe, minimal_roi, stoploss, trailing_* """ # Strategy interface version - allow new iterations of the strategy interface. # Check the documentation or the Sample strategy to get the latest version. INTERFACE_VERSION = 3 # Optimal timeframe for the strategy. timeframe = "1m" # Can this strategy go short? can_short: bool = True # Minimal ROI designed for the strategy. # This attribute will be overridden if the config file contains "minimal_roi". minimal_roi = { "0": 100000 } # Optimal stoploss designed for the strategy. # This attribute will be overridden if the config file contains "stoploss". stoploss = -100000 # Trailing stoploss # trailing_stop = False # trailing_only_offset_is_reached = False # trailing_stop_positive = 0.01 # trailing_stop_positive_offset = 0.0 # Disabled / not configured # Run "populate_indicators()" only for new candle. process_only_new_candles = True # These values can be overridden in the config. use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 30 # Strategy parameters # buy_rsi = IntParameter(10, 40, default=30, space="buy") # sell_rsi = IntParameter(60, 90, default=70, space="sell")# Optional order type mapping. order_types = { "entry": "limit", "exit": "limit", "stoploss": "market", "stoploss_on_exchange": False } # Optional order time in force. order_time_in_force = { "entry": "GTC", "exit": "GTC" } @property def plot_config(self): return { # Main plot indicators (Moving averages, ...) "main_plot": { "tema": {}, "sar": {"color": "white"}, }, "subplots": { # Subplots - each dict defines one additional plot "MACD": { "macd": {"color": "blue"}, "macdsignal": {"color": "orange"}, }, "RSI": { "rsi": {"color": "red"}, } } } def informative_pairs(self): """ Define additional, informative pair/interval combinations to be cached from the exchange. These pair/interval combinations are non-tradeable, unless they are part of the whitelist as well. For more information, please consult the documentation :return: List of tuples in the format (pair, interval) Sample: return [("ETH/USDT", "5m"), ("BTC/USDT", "15m"), ] """ return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict): dataframe = dataframe[-100:] # 计算5分钟涨跌幅 recent = dataframe.iloc[-5:] # 最近5分钟 open_price = recent.iloc[0]['open'] close_price = recent.iloc[-1]['close'] pct_change = (close_price - open_price) / open_price dataframe.loc[dataframe.index[-1], "price_change"] = pct_change # 计算5分钟成交量 dataframe["5m_volume"] = dataframe["volume"].rolling(window=5).sum() windows = 30 mean_vol = dataframe['5m_volume'].rolling(window=windows).mean() std_vol = dataframe['5m_volume'].rolling(window=windows).std() dataframe["z_score"] = (dataframe['5m_volume'] - mean_vol) / std_vol return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict): dataframe = dataframe[-100:] dataframe.loc[ ( (dataframe['price_change'] > 0.01) & # 价格上涨 (dataframe['z_score'] > 2) # Z-score大于1 ), 'enter_long'] = 1 dataframe.loc[ ( (dataframe['price_change'] < -0.01) & # 价格下跌 (dataframe['z_score'] > 2) # Z-score小于-1 ), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict): dataframe = dataframe[-100:] return dataframe