IchiV2_LS_Live - for live config bots
Timeframe
1h
Direction
Long & Short
Stoploss
-12.9%
Trailing Stop
No
ROI
0m: 1000.0%
Interface Version
3
Startup Candles
121
Indicators
5
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# IchiV2_LS_1h_Merged - Dual Long/Short Ichimoku Strategy (1h timeframe)
# Author: Gui
# Version: 2.0
# Description: Combines cloud crossover (long) with senkou B retest breakdown (short)
from freqtrade.strategy.interface import IStrategy
from pandas import DataFrame
import talib.abstract as ta
import pandas as pd
pd.options.mode.chained_assignment = None
import technical.indicators as ftt
import freqtrade.vendor.qtpylib.indicators as qtpylib
from datetime import datetime
from freqtrade.persistence import Trade
from freqtrade.enums import RunMode
from freqtrade.strategy import informative, DecimalParameter, IntParameter
from freqtrade.strategy.strategy_helper import stoploss_from_absolute
from freqtrade.exchange import timeframe_to_prev_date
import logging
import json
from freqtrade.optimize.space import SKDecimal
class IchiV2_LS_Live(IStrategy):
"""
IchiV2_LS_Live - for live config bots
Dual long/short Ichimoku strategy on 1h timeframe.
Long entries (checked at 4h boundary):
- 4h close crosses above senkou_a_4h
- Cloud is green (senkou_a_4h > senkou_b_4h)
- SAR is below close (sar_4h < close_4h)
Long exits:
- 1h close crosses below sar_4h (exit on SAR cross)
- Base stoploss at -0.2 as fallback
- No ATR stop
Short entries (checked on any 1h candle):
- Cloud is bearish (senkou_a < senkou_b)
- Price retested senkou_b (high >= senkou_b) within lookback window
- Current price breaks down below senkou_b
Short exits:
- RSI 4h oversold (< 25)
- OR price velocity exhaustion (< -10%)
- ATR-based stop loss (3.5x ATR from entry, locked at entry)
- No SAR exit for shorts
Standard Freqtrade position sizing (no market cap scaling).
MIXED TAG STATS
┏━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ ┃ ┃ ┃ ┃ Tot ┃ ┃ ┃ ┃
┃ ┃ Exit ┃ ┃ Avg Profit ┃ Profit ┃ Tot Profit ┃ Avg ┃ Win Draw ┃
┃ Enter Tag ┃ Reason ┃ Trades ┃ % ┃ USDT ┃ % ┃ Duration ┃ Loss Win% ┃
┡━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ short_en… │ tp_veloc… │ 184 │ 11.93 │ 702.255 │ 702.25 │ 2 days, │ 168 0 │
│ │ │ │ │ │ │ 11:11:00 │ 16 91.3 │
│ long_ent… │ sar_exit… │ 1121 │ 2.38 │ 606.182 │ 606.18 │ 1 day, │ 515 0 │
│ │ │ │ │ │ │ 21:13:00 │ 606 45.9 │
│ short_en… │ tp_rsi_4… │ 159 │ 13.38 │ 596.850 │ 596.85 │ 6 days, │ 143 0 │
│ │ │ │ │ │ │ 15:31:00 │ 16 89.9 │
│ short_en… │ senkou_a… │ 431 │ 2.0 │ 288.346 │ 288.35 │ 4 days, │ 299 0 │
│ │ │ │ │ │ │ 3:47:00 │ 132 69.4 │
│ short_en… │ force_ex… │ 3 │ 5.33 │ 15.784 │ 15.78 │ 5 days, │ 3 0 │
│ │ │ │ │ │ │ 19:20:00 │ 0 100 │
│ long_ent… │ senkou_a… │ 103 │ -0.25 │ -10.824 │ -10.82 │ 19:07:00 │ 42 0 │
│ │ │ │ │ │ │ │ 61 40.8 │
│ short_en… │ trailing… │ 257 │ -3.61 │ -220.982 │ -220.98 │ 1 day, │ 2 0 │
│ │ │ │ │ │ │ 23:23:00 │ 255 0.8 │
│ long_ent… │ stop_loss │ 186 │ -7.23 │ -328.957 │ -328.96 │ 16:19:00 │ 0 0 │
│ │ │ │ │ │ │ │ 186 0 │
│ short_en… │ stop_loss │ 651 │ -4.13 │ -709.723 │ -709.72 │ 1 day, │ 0 0 │
│ │ │ │ │ │ │ 12:31:00 │ 651 0 │
│ TOTAL │ │ 3095 │ 0.93 │ 938.931 │ 938.93 │ 2 days, │ 1172 0 │
│ │ │ │ │ │ │ 7:21:00 │ 1923 37.9 │
└───────────┴───────────┴────────┴────────────┴───────────┴────────────┴───────────┴────────────┘
SUMMARY METRICS
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Metric ┃ Value ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ Backtesting from │ 2020-10-21 20:00:00 │
│ Backtesting to │ 2025-12-24 00:00:00 │
│ Trading Mode │ Isolated Futures │
│ Max open trades │ 10 │
│ │ │
│ Total/Daily Avg Trades │ 3095 / 1.64 │
│ Starting balance │ 100 USDT │
│ Final balance │ 1038.931 USDT │
│ Absolute profit │ 938.931 USDT │
│ Total profit % │ 938.93% │
│ CAGR % │ 57.19% │
│ Sortino │ 3.25 │
│ Sharpe │ 1.59 │
│ Calmar │ 5.32 │
│ SQN │ 4.99 │
│ Profit factor │ 1.51 │
│ Expectancy (Ratio) │ 0.30 (0.32) │
│ Avg. daily profit │ 0.497 USDT │
│ Avg. stake amount │ 26.458 USDT │
│ Total trade volume │ 163397.423 USDT │
│ │ │
│ Long / Short trades │ 1410 / 1685 │
│ Long / Short profit % │ 266.40% / 672.53% │
│ Long / Short profit USDT │ 266.400 / 672.530 │
│ │ │
│ Best Pair │ MOODENG/USDT:USDT 118.52% │
│ Worst Pair │ BNB/USDT:USDT -15.69% │
│ Best trade │ MOODENG/USDT:USDT 196.35% │
│ Worst trade │ NEAR/USDT:USDT -26.36% │
│ Best day │ 100.93 USDT │
│ Worst day │ -28.59 USDT │
│ Days win/draw/lose │ 469 / 658 / 763 │
│ Min/Max/Avg. Duration Winners │ 0d 00:00 / 30d 23:00 / 3d 17:15 │
│ Min/Max/Avg. Duration Losers │ 0d 00:00 / 17d 09:00 / 1d 10:42 │
│ Max Consecutive Wins / Loss │ 17 / 24 │
│ Rejected Entry signals │ 694 │
│ Entry/Exit Timeouts │ 0 / 0 │
│ │ │
│ Min balance │ 91.397 USDT │
│ Max balance │ 1038.931 USDT │
│ Max % of account underwater │ 19.66% │
│ Absolute drawdown │ 89.8 USDT (10.74%) │
│ Drawdown duration │ 50 days 01:00:00 │
│ Profit at drawdown start │ 736.105 USDT │
│ Profit at drawdown end │ 646.306 USDT │
│ Drawdown start │ 2025-08-17 16:00:00 │
│ Drawdown end │ 2025-10-06 17:00:00 │
│ Market change │ 353.16% │
└───────────────────────────────┴─────────────────────────────────┘
Backtested 2020-10-21 20:00:00 -> 2025-12-24 00:00:00 | Max open trades : 10
STRATEGY SUMMARY
┏━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┓
┃ ┃ ┃ ┃ Tot ┃ ┃ ┃ ┃ ┃
┃ ┃ ┃ Avg Profit ┃ Profit ┃ Tot Profit ┃ Avg ┃ Win Draw ┃ ┃
┃ Strategy ┃ Trades ┃ % ┃ USDT ┃ % ┃ Duration ┃ Loss Win% ┃ Drawdown ┃
┡━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━┩
│ IchiV2_L… │ 3095 │ 0.93 │ 938.931 │ 938.93 │ 2 days, │ 1172 0 │ 89.8 USDT │
│ │ │ │ │ │ 7:21:00 │ 1923 37.9 │ 10.74% │
└───────────┴────────┴────────────┴───────────┴────────────┴───────────┴────────────┴───────────┘
➜
"""
def __init__(self, config):
super().__init__(config)
self.logger = logging.getLogger(__name__)
self._pairs_analyzed_count = 0
def _log_jsonl(self, signal_type: str, pair: str = None,
conditions: dict = None, indicators: dict = None,
trade_data: dict = None, **extra_kwargs):
"""
Universal JSONL logger for strategy analysis.
:param signal_type: Type of signal/analysis (e.g., 'long_entry_check', 'short_exit_check')
:param pair: Trading pair
:param conditions: Dict of conditions being evaluated {name: bool}
:param indicators: Dict of indicator values {name: float}
:param trade_data: Dict of trade-specific data (trade_id, profit, etc.)
:param extra_kwargs: Any additional key-value pairs to log
"""
import numpy as np
def convert_to_serializable(obj, max_decimals=4):
"""Convert numpy/pandas types to native Python types with decimal precision limit."""
if isinstance(obj, (np.integer, np.floating)):
return round(float(obj), max_decimals)
elif isinstance(obj, np.bool_):
return bool(obj)
elif isinstance(obj, float):
return round(obj, max_decimals)
elif isinstance(obj, dict):
return {k: convert_to_serializable(v, max_decimals) for k, v in obj.items()}
elif isinstance(obj, (list, tuple)):
return [convert_to_serializable(item, max_decimals) for item in obj]
elif pd.isna(obj):
return None
return obj
log_entry = {
"timestamp": datetime.now().isoformat(),
"signal_type": signal_type,
}
# Add pair if provided
if pair:
log_entry["pair"] = pair
# Add conditions if provided
if conditions:
log_entry["conditions"] = convert_to_serializable(conditions)
# Add condition summary
log_entry["conditions_passed"] = sum(1 for v in conditions.values() if v)
log_entry["conditions_total"] = len(conditions)
log_entry["all_conditions_met"] = all(conditions.values())
# Add indicators if provided
if indicators:
log_entry["indicators"] = convert_to_serializable(indicators)
# Add trade data if provided
if trade_data:
log_entry["trade"] = convert_to_serializable(trade_data)
# Add any extra kwargs
log_entry.update(convert_to_serializable(extra_kwargs))
# Log as single-line JSON (no prefix for easy parsing)
self.logger.info(json.dumps(log_entry))
def _calculate_atr_stop_price(self, current_price: float, atr: float, is_short: bool) -> dict:
"""
Calculate ATR-based stop loss price and related metrics.
:param current_price: Current market price
:param atr: Current ATR value
:param is_short: True if short position, False if long
:return: Dict with stop price and distance info
"""
atr_multiplier = self.atr_stop_multiplier.value
atr_distance = atr_multiplier * atr
if is_short:
# Short: stop is ABOVE entry (price + distance)
stop_price = current_price + atr_distance
stop_pct = (atr_distance / current_price) * 100
else:
# Long: stop is BELOW entry (price - distance)
stop_price = current_price - atr_distance
stop_pct = (atr_distance / current_price) * 100
return {
'atr_stop_price': stop_price,
'atr_stop_distance': atr_distance,
'atr_stop_pct': stop_pct,
'atr_multiplier': atr_multiplier,
}
INTERFACE_VERSION = 3
can_short = True # Set to False to test long-only performance
# Hyperoptable parameters
buy_params = {
# Short entry (RetestSenkou)
'retest_lookback_window': 24, # Max candles to look for retest (24h)
'rsi_tp_threshold': 25, # RSI oversold threshold for shorts
'velocity_tp_threshold': -0.1, # Velocity threshold for shorts
'price_velocity_window': 10, # Velocity calculation window
}
# Short entry parameters
retest_lookback_window = IntParameter(4, 96, default=24, space='buy', optimize=False)
# Short exit parameters
rsi_tp_threshold = IntParameter(15, 35, default=25, space='sell', optimize=False)
use_rsi_exit = IntParameter(0, 1, default=1, space='sell', optimize=False)
# ATR stop loss for shorts
atr_stop_multiplier = DecimalParameter(1.5, 4.0, default=3.5, decimals=1, space='sell', optimize=False)
atr_timeperiod = IntParameter(2, 20, default=14, space='sell', optimize=False)
# Price velocity window
velocity_tp_threshold = DecimalParameter(-0.13, -0.06, default=-0.1, decimals=2, space='sell', optimize=True)
price_velocity_window = IntParameter(4, 20, default=20, space='sell', optimize=False)
sell_params = {}
custom_info = {}
# ROI table - minimal
minimal_roi = {
"0": 10.0 # Disabled
}
# Stop loss
# -0.2 for longs (only SAR exit matters)
# Shorts use ATR-based custom stoploss
stoploss = -0.129
use_custom_stoploss = True # Only applies to shorts (returns None for longs)
# Timeframe
timeframe = '1h'
startup_candle_count = 121
process_only_new_candles = True
# Trailing stop - DISABLED
trailing_stop = False
trailing_stop_positive = None
trailing_stop_positive_offset = None
trailing_only_offset_is_reached = False
# Exit settings
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = True
ICHI_PARAMS = dict(
conversion_line_period=20,
base_line_periods=60,
laggin_span=120,
displacement=1,
)
plot_config = {
'main_plot': {
'sar_4h': {'color': 'purple', 'marker': '.', 'linestyle': 'none'},
'senkou_a_4h': {'color': 'lightgreen', 'linestyle': '--'},
'senkou_b_4h': {'color': 'lightcoral', 'linestyle': '--'},
# 4h Bollinger Bands
# 'bb_upper_4h': {'color': 'lightblue'},
# 'bb_middle_4h': {'color': 'lightgray'},
# 'bb_lower_4h': {'color': 'lightblue'},
},
'subplots': {
'RSI 4h': {
'rsi_4h': {'color': 'purple'},
},
'Price Velocity': {
'price_velocity': {'color': 'magenta'},
},
},
}
def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, after_fill: bool,
**kwargs) -> float | None:
"""
Dynamic stoploss comparing base stoploss vs strategy-specific stops.
LONGS: Compare base stoploss vs SAR level, choose the TIGHTER (higher) stop.
SHORTS: Compare base stoploss vs ATR stop, choose the TIGHTER (lower) stop.
"""
if not trade.is_short:
# === LONG POSITIONS: Compare base stoploss vs SAR ===
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe.empty:
return None
try:
last_candle = dataframe.iloc[-1]
sar_level = last_candle.get('sar_4h')
if pd.isna(sar_level):
return None
# Calculate base stoploss price (below entry)
entry_rate = trade.open_rate
base_stop_price = entry_rate * (1 + self.stoploss) # self.stoploss is negative
# For longs: Choose the HIGHER stop price (closer to current price = more conservative)
# SAR trails up as price goes up, so it may be higher than base stoploss
final_stop_price = max(base_stop_price, sar_level)
return stoploss_from_absolute(
stop_rate=final_stop_price,
current_rate=current_rate,
is_short=False,
leverage=trade.leverage
)
except (KeyError, IndexError) as e:
self.logger.warning(f"Error calculating SAR stoploss for {pair}: {e}")
return None
else:
# === SHORT POSITIONS: Compare base stoploss vs ATR stop ===
entry_atr = trade.get_custom_data(key='entry_atr')
if entry_atr is None:
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if not dataframe.empty:
try:
trade_candle_time = timeframe_to_prev_date(self.timeframe, trade.open_date_utc)
entry_candles = dataframe[dataframe['date'] <= trade_candle_time]
if not entry_candles.empty:
entry_atr = entry_candles.iloc[-1]['atr']
if not pd.isna(entry_atr):
trade.set_custom_data(key='entry_atr', value=float(entry_atr))
except (KeyError, IndexError) as e:
self.logger.warning(f"Failed to store entry ATR for {pair}: {e}")
if entry_atr is None:
return None
entry_rate = trade.open_rate
atr_distance = self.atr_stop_multiplier.value * float(entry_atr)
# For shorts: ATR stop is ABOVE entry price
atr_stop_price = entry_rate + atr_distance
base_stop_price = entry_rate * (1 + abs(self.stoploss))
# For shorts: Choose the LOWER stop price (closer to current price = more conservative)
final_stop_price = min(atr_stop_price, base_stop_price)
return stoploss_from_absolute(
stop_rate=final_stop_price,
current_rate=current_rate,
is_short=True,
leverage=trade.leverage
)
def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> str | bool | None:
"""
Exit logic - COMPLETELY SEPARATE for longs vs shorts:
LONGS (from IchiV3_LS_1h):
- SAR exit: 1h close crosses below sar_4h → exit immediately
- This is the PRIMARY and ONLY exit for longs (base stoploss is just fallback)
SHORTS (from IchiV2_Short_RetestSenkou):
- RSI 4h oversold (< 25)
- OR velocity exhaustion (< -10%)
- ATR-based stop loss (via custom_stoploss)
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe.empty:
return None
current_mode = self.config['runmode']
is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN)
try:
last_row = dataframe.iloc[-1]
last_close = last_row['close']
last_sar = last_row.get('sar_4h')
# LONG EXIT: SAR
if not trade.is_short:
# Cloud flip check (to prevent concurrent trades between long and short)
cloud_flip = last_row['senkou_a_4h'] < last_row['senkou_b_4h']
if cloud_flip:
if is_trading_mode:
self._log_jsonl(
signal_type="long_exit_check",
pair=pair,
conditions={'cloud_flip_bearish': cloud_flip},
indicators={
'close': float(last_close),
'sar_4h': float(last_sar) if not pd.isna(last_sar) else None,
'senkou_a_4h': float(last_row['senkou_a_4h']),
'senkou_b_4h': float(last_row['senkou_b_4h']),
},
trade_data={
'trade_id': trade.id,
'current_profit': current_profit,
'current_profit_pct': current_profit * 100,
},
exit_triggered=True,
exit_reason='senkou_a_4h_exit_long'
)
return 'senkou_a_4h_exit_long'
# SAR cross down
sar_cross_down = not pd.isna(last_sar) and last_close <= last_sar
if is_trading_mode:
self._log_jsonl(
signal_type="long_exit_check",
pair=pair,
conditions={'sar_cross_down': sar_cross_down},
indicators={
'close': float(last_close),
'sar_4h': float(last_sar) if not pd.isna(last_sar) else None,
},
trade_data={
'trade_id': trade.id,
'current_profit': current_profit,
'current_profit_pct': current_profit * 100,
},
exit_triggered=sar_cross_down,
exit_reason='sar_exit_long' if sar_cross_down else None
)
if sar_cross_down:
return 'sar_exit_long'
# SHORT EXIT: RSI + Velocity
if trade.is_short:
last_rsi_4h = last_row.get('rsi_4h')
last_velocity = last_row.get('price_velocity')
# Cloud flip check (to prevent concurrent trades between long and short)
cloud_flip = last_row['senkou_a_4h'] > last_row['senkou_b_4h']
if cloud_flip:
if is_trading_mode:
self._log_jsonl(
signal_type="short_exit_check",
pair=pair,
conditions={'cloud_flip_bullish': cloud_flip},
indicators={
'close': float(last_close),
'senkou_a_4h': float(last_row['senkou_a_4h']),
'senkou_b_4h': float(last_row['senkou_b_4h']),
'rsi_4h': float(last_rsi_4h) if not pd.isna(last_rsi_4h) else None,
'price_velocity': float(last_velocity) if not pd.isna(last_velocity) else None,
'atr': float(last_row.get('atr', 0)),
},
trade_data={
'trade_id': trade.id,
'current_profit': current_profit,
'current_profit_pct': current_profit * 100,
},
exit_triggered=True,
exit_reason='senkou_a_4h_exit_short'
)
return 'senkou_a_4h_exit_short'
# Check exit conditions
rsi_oversold = False
velocity_exhausted = False
if self.use_rsi_exit.value == 1:
rsi_oversold = not pd.isna(last_rsi_4h) and last_rsi_4h < self.rsi_tp_threshold.value
velocity_exhausted = not pd.isna(last_velocity) and last_velocity < self.velocity_tp_threshold.value
if is_trading_mode:
self._log_jsonl(
signal_type="short_exit_check",
pair=pair,
conditions={
'rsi_oversold': rsi_oversold,
'velocity_exhausted': velocity_exhausted,
},
indicators={
'close': float(last_close),
'rsi_4h': float(last_rsi_4h) if not pd.isna(last_rsi_4h) else None,
'price_velocity': float(last_velocity) if not pd.isna(last_velocity) else None,
'atr': float(last_row.get('atr', 0)),
'rsi_threshold': self.rsi_tp_threshold.value,
'velocity_threshold': self.velocity_tp_threshold.value,
},
trade_data={
'trade_id': trade.id,
'current_profit': current_profit,
'current_profit_pct': current_profit * 100,
},
exit_triggered=rsi_oversold or velocity_exhausted,
exit_reason='tp_rsi_4h_oversold' if rsi_oversold else ('tp_velocity_achieved' if velocity_exhausted else None)
)
if rsi_oversold:
return 'tp_rsi_4h_oversold'
if velocity_exhausted:
return 'tp_velocity_achieved'
except (KeyError, IndexError) as e:
self.logger.warning(f"Error in custom_exit for {pair}: {e}")
return None
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: str | None,
side: str, **kwargs) -> bool:
"""Store ATR at entry time for shorts and log detailed entry information."""
trade = kwargs.get('trade')
# Store ATR for shorts
if trade is not None and side == 'short':
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if not dataframe.empty:
try:
entry_candle = dataframe.iloc[-1]
entry_atr = entry_candle.get('atr')
if not pd.isna(entry_atr):
trade.set_custom_data(key='entry_atr', value=float(entry_atr))
except (KeyError, IndexError) as e:
self.logger.warning(f"Failed to store entry ATR for {pair}: {e}")
current_mode = self.config['runmode']
is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN)
if is_trading_mode:
self.logger.info(f"ENTRY | {pair} | {side} | amount={amount} | rate={rate} | time={current_time}")
return True
def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float,
rate: float, time_in_force: str, exit_reason: str,
current_time: datetime, **kwargs) -> bool:
"""Log detailed trade exit information."""
current_mode = self.config['runmode']
is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN)
if is_trading_mode:
profit_ratio = trade.calc_profit_ratio(rate)
duration = current_time - trade.open_date_utc
side = 'short' if trade.is_short else 'long'
self.logger.info(f"EXIT | {pair} | {side} | reason={exit_reason} | profit={profit_ratio:.2%} | duration={duration} | time={current_time}")
return True
def bot_loop_start(self, current_time: datetime, **kwargs) -> None:
"""
Called at the start of each trading cycle.
Logs balance and open positions summary.
"""
current_mode = self.config['runmode']
is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN)
if is_trading_mode:
self.logger.info(f"🔄 NEW TRADING CYCLE: {current_time}")
self._log_portfolio_summary(current_time, **kwargs)
# Reset counter for new cycle
self._pairs_analyzed_count = 0
def _log_portfolio_summary(self, cycle_time: datetime, **kwargs) -> None:
"""
Log portfolio summary with balance and open positions details in JSONL format.
"""
open_trades = Trade.get_open_trades()
# Get balance information
balance_info = {}
stake_currency = self.config.get('stake_currency', 'USDC')
wallets = kwargs.get('wallets') or self.wallets
if wallets:
try:
balance_info = {
'total_balance': float(wallets.get_total(stake_currency)),
'total_stake_amount': float(wallets.get_total_stake_amount()),
'free_balance': float(wallets.get_free(stake_currency)),
'used_balance': float(wallets.get_used(stake_currency)),
'stake_currency': stake_currency,
}
except Exception as e:
self.logger.warning(f"Failed to fetch balance info: {e}")
balance_info = {'error': str(e)}
# Collect position details
positions = []
total_unrealized_pnl = 0
winning = 0
losing = 0
for trade in open_trades:
# Get current rate from dataprovider
try:
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
if not dataframe.empty:
current_rate = dataframe.iloc[-1]['close']
else:
current_rate = trade.open_rate
except Exception:
current_rate = trade.open_rate
profit_ratio = trade.calc_profit_ratio(current_rate)
profit_pct = profit_ratio * 100
entry_date = trade.open_date_utc
duration = cycle_time - entry_date
duration_seconds = duration.total_seconds()
total_unrealized_pnl += profit_ratio
if profit_ratio > 0:
winning += 1
else:
losing += 1
side_label = "short" if trade.is_short else "long"
positions.append({
'trade_id': trade.id,
'pair': trade.pair,
'side': side_label,
'entry_rate': float(trade.open_rate),
'current_rate': float(current_rate),
'entry_time': entry_date.isoformat(),
'profit_pct': float(profit_pct),
'profit_ratio': float(profit_ratio),
'duration_seconds': int(duration_seconds),
'duration_str': str(duration),
'is_winning': profit_ratio > 0,
})
# Log portfolio summary as JSONL
self._log_jsonl(
signal_type="portfolio_summary",
cycle_time=cycle_time.isoformat(),
balance=balance_info,
positions_count=len(open_trades),
positions=positions,
summary={
'winning_positions': winning,
'losing_positions': losing,
'total_unrealized_pnl_pct': float(total_unrealized_pnl * 100),
'total_unrealized_pnl_ratio': float(total_unrealized_pnl),
}
)
def _add_ichimoku(self, df: DataFrame) -> None:
ichi = ftt.ichimoku(df, **self.ICHI_PARAMS)
df['senkou_a'] = ichi['senkou_span_a']
df['senkou_b'] = ichi['senkou_span_b']
df['chikou_span'] = ichi['chikou_span']
df['tenkan_sen'] = ichi['tenkan_sen']
df['kijun_sen'] = ichi['kijun_sen']
df['cloud_green'] = ichi['cloud_green']
df['cloud_red'] = ichi['cloud_red']
@informative('4h')
def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Calculate 4h Ichimoku, SAR, and RSI."""
self._add_ichimoku(dataframe)
dataframe['sar'] = ta.SAR(dataframe)
dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14)
# # Bollinger Bands for analysis (using qtpylib)
# bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=3.0)
# dataframe['bb_upper'] = bollinger['upper']
# dataframe['bb_middle'] = bollinger['mid']
# dataframe['bb_lower'] = bollinger['lower']
return dataframe
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Calculate 1h indicators."""
# Mark 4h boundaries for long entries
dataframe['is_4h_boundary'] = dataframe['date'].dt.hour % 4 == 0
# Cloud status for short entries
dataframe['cloud_bearish'] = dataframe['senkou_a_4h'] < dataframe['senkou_b_4h']
# Price velocity for short exits
dataframe['price_velocity'] = (
(dataframe['close'] - dataframe['close'].shift(self.price_velocity_window.value)) /
dataframe['close'].shift(self.price_velocity_window.value)
)
# ATR for short stop loss
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
# Store for custom functions
self.custom_info[metadata['pair']] = dataframe[[
'date', 'close', 'sar_4h', 'rsi_4h', 'price_velocity', 'atr',
'senkou_a_4h', 'senkou_b_4h'
# 'bb_upper_4h'
]].copy().set_index('date')
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Dual entry system:
1. LONG: Ichimoku cloud crossover at 4h boundaries
2. SHORT: Senkou B retest breakdown
"""
pair = metadata.get('pair', 'unknown')
current_mode = self.config['runmode']
is_trading_mode = current_mode in (RunMode.LIVE, RunMode.DRY_RUN)
# Initialize
dataframe['enter_long'] = 0
dataframe['enter_short'] = 0
dataframe['buy'] = 0
dataframe['enter_tag'] = ''
# === LONG ENTRY (IchiV3_LS_1h logic) ===
# Only at 4h boundaries
at_4h_boundary = dataframe['is_4h_boundary']
effective_close = dataframe['close_4h']
# Long: prev 4h close was below senkou_a, current 4h close is above
prev_close_4h = effective_close.shift(4)
prev_senkou_a = dataframe['senkou_a_4h'].shift(4)
prev_below_cloud = prev_close_4h < prev_senkou_a
curr_above_cloud = effective_close > dataframe['senkou_a_4h']
cloud_green = dataframe['senkou_a_4h'] > dataframe['senkou_b_4h']
sar_below_close = dataframe['sar_4h'] < effective_close
long_entry = at_4h_boundary & prev_below_cloud & curr_above_cloud & cloud_green & sar_below_close
# === SHORT ENTRY (RetestSenkou logic) ===
window = self.retest_lookback_window.value
# Retest detection
dataframe['retested_senkou_b'] = dataframe['high'] >= dataframe['senkou_b_4h']
dataframe['had_senkou_b_retest'] = dataframe['retested_senkou_b'].rolling(
window=window, min_periods=1
).max().astype(bool)
# Breakdown detection
dataframe['cross_down_senkou_b'] = (
(dataframe['close'].shift(1) >= dataframe['senkou_b_4h'].shift(1)) &
(dataframe['close'] < dataframe['senkou_b_4h'])
)
# Short entry: cloud bearish + retest + breakdown
short_entry = (
dataframe['cloud_bearish'] &
dataframe['had_senkou_b_retest'] &
dataframe['cross_down_senkou_b']
)
# Apply signals
dataframe.loc[long_entry, 'enter_long'] = 1
dataframe.loc[long_entry, 'enter_tag'] = 'long_entry'
dataframe.loc[short_entry, 'enter_short'] = 1
dataframe.loc[short_entry, 'enter_tag'] = 'short_entry'
dataframe.loc[long_entry | short_entry, 'buy'] = 1
# Signal logging (only in trading mode)
if is_trading_mode and len(dataframe) > 0:
latest = dataframe.iloc[-1]
# LONG signal analysis
if latest['is_4h_boundary']:
long_conditions = {
'prev_below_cloud': bool(prev_below_cloud.iloc[-1]) if len(prev_below_cloud) > 0 else False,
'curr_above_cloud': bool(curr_above_cloud.iloc[-1]) if len(curr_above_cloud) > 0 else False,
'cloud_green': bool(cloud_green.iloc[-1]) if len(cloud_green) > 0 else False,
'sar_below_close': bool(sar_below_close.iloc[-1]) if len(sar_below_close) > 0 else False,
}
long_signal = latest['enter_long'] == 1
# Calculate ATR stop for longs (informational - longs use SAR exit)
atr_stop_info = self._calculate_atr_stop_price(
current_price=float(latest['close']),
atr=float(latest.get('atr', 0)),
is_short=False
)
# Standardized JSONL logging
self._log_jsonl(
signal_type="long_entry_check",
pair=pair,
conditions=long_conditions,
indicators={
'close': float(latest['close']),
'close_4h': float(latest['close_4h']),
'senkou_a_4h': float(latest['senkou_a_4h']),
'senkou_b_4h': float(latest['senkou_b_4h']),
'sar_4h': float(latest['sar_4h']),
'rsi_4h': float(latest.get('rsi_4h', 0)),
'atr': float(latest.get('atr', 0)),
'atr_stop_price': atr_stop_info['atr_stop_price'],
'atr_stop_distance': atr_stop_info['atr_stop_distance'],
'atr_stop_pct': atr_stop_info['atr_stop_pct'],
},
signal_triggered=long_signal,
is_4h_boundary=True,
action='ENTER_LONG' if long_signal else 'REJECTED',
note='longs_use_sar_exit_not_atr'
)
# SHORT signal analysis
short_conditions = {
'cloud_bearish': bool(latest['cloud_bearish']),
'had_retest': bool(latest['had_senkou_b_retest']),
'cross_down': bool(latest['cross_down_senkou_b']),
}
short_signal = latest['enter_short'] == 1
# Only log if at least one condition is met or signal triggered (reduce noise)
should_log_short = short_signal or latest['cloud_bearish'] or latest['had_senkou_b_retest'] or latest['cross_down_senkou_b']
if should_log_short:
# Calculate ATR stop for shorts (this is the actual stop used)
atr_stop_info = self._calculate_atr_stop_price(
current_price=float(latest['close']),
atr=float(latest.get('atr', 0)),
is_short=True
)
# Standardized JSONL logging
self._log_jsonl(
signal_type="short_entry_check",
pair=pair,
conditions=short_conditions,
indicators={
'close': float(latest['close']),
'senkou_a_4h': float(latest['senkou_a_4h']),
'senkou_b_4h': float(latest['senkou_b_4h']),
'rsi_4h': float(latest.get('rsi_4h', 0)),
'price_velocity': float(latest.get('price_velocity', 0)),
'atr': float(latest.get('atr', 0)),
'atr_stop_price': atr_stop_info['atr_stop_price'],
'atr_stop_distance': atr_stop_info['atr_stop_distance'],
'atr_stop_pct': atr_stop_info['atr_stop_pct'],
},
signal_triggered=short_signal,
action='ENTER_SHORT' if short_signal else 'REJECTED',
note='shorts_use_atr_stop'
)
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Exit signals disabled - using custom_exit."""
dataframe['exit_long'] = 0
dataframe['exit_short'] = 0
dataframe['exit_tag'] = ''
dataframe['sell'] = 0
return dataframe