TopGun Trading Strategy All indicators ported exactly from Pine Script.
Timeframe
4h
Direction
Long Only
Stoploss
-3.0%
Trailing Stop
Yes
ROI
0m: 6.0%, 60m: 4.0%, 120m: 2.0%, 240m: 1.0%
Interface Version
3
Startup Candles
N/A
Indicators
5
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
"""
╔══════════════════════════════════════════════════════════════════════╗
║ TopGun Trading Strategy — Freqtrade ║
║ ║
║ Indicators (exact Pine Script ports): ║
║ 1. Pi Cycle High/Low (cycle top & bottom detection) ║
║ 2. [RS] JMA Modified RSI (3-length: 10 / 60 / 420) ║
║ 3. TTM Squeeze (BB vs KC momentum) ║
║ 4. RSI + Pivot Divergence (bull/bear labels) ║
║ 5. Kalit MA Stack (MA5/10/20/50/100/200 + RSI2 rules) ║
║ ║
║ Timeframe : 4h (momentum) with 1d informative (trend) ║
║ Pairs : BTC/USDT ETH/USDT SOL/USDT ║
╚══════════════════════════════════════════════════════════════════════╝
"""
from datetime import datetime
from typing import Optional
import numpy as np
import pandas as pd
from pandas import DataFrame
from freqtrade.strategy import IStrategy, informative
import talib.abstract as ta
class TopGunStrategy(IStrategy):
"""
TopGun Trading Strategy
All indicators ported exactly from Pine Script.
"""
# ── Strategy metadata ────────────────────────────────────────────
INTERFACE_VERSION = 3
timeframe = "4h"
can_short = False
# ── ROI & Stoploss ───────────────────────────────────────────────
minimal_roi = {
"0": 0.06, # 6% take profit immediately
"60": 0.04, # 4% after 60 min
"120": 0.02, # 2% after 120 min
"240": 0.01, # 1% after 240 min
}
stoploss = -0.03 # 3% stop loss
# ── Trailing stop ────────────────────────────────────────────────
trailing_stop = True
trailing_stop_positive = 0.02
trailing_stop_positive_offset = 0.04
trailing_only_offset_is_reached = True
# ── Misc ─────────────────────────────────────────────────────────
startup_candle_count: int = 500 # need lots of history for MA300/Pi Cycle
process_only_new_candles = True
# ══════════════════════════════════════════════════════════════════
# HELPER FUNCTIONS
# ══════════════════════════════════════════════════════════════════
def _rma(self, series: pd.Series, length: int) -> pd.Series:
"""Wilder's RMA — same as Pine ta.rma"""
return series.ewm(alpha=1 / length, adjust=False).mean()
def _jmax(self, src: np.ndarray, length: int, exp: float) -> np.ndarray:
"""
Exact port of capissimo's jmax() Pine Script function.
Jurik Moving Average core calculation.
"""
beta = 0.45 * (length - 1) / (0.45 * (length - 1) + 2)
alpha = beta ** exp
n = len(src)
L0 = L1 = L3 = L4 = 0.0
out = np.full(n, np.nan)
for i in range(n):
v = src[i]
if np.isnan(v):
continue
L0 = (1 - alpha) * v + alpha * L0
L1 = (v - L0) * (1 - beta) + beta * L1
L2 = L0 + L1
L3 = (L2 - L4) * ((1 - alpha) ** 2) + (alpha ** 2) * L3
L4 = L4 + L3
out[i] = L4
return out
def _jma_mod_rsi(self, src: np.ndarray, length: int, exp: float = 2.0) -> np.ndarray:
"""
[RS] JMA Modified RSI — exact Pine Script port.
RSI where the average is replaced by JMA instead of a simple mean.
"""
avg = self._jmax(src, length, exp)
u = np.maximum(src - avg, 0)
d = np.maximum(avg - src, 0)
ru = self._rma(pd.Series(u), length).values
rd = self._rma(pd.Series(d), length).values
with np.errstate(divide="ignore", invalid="ignore"):
rs = np.where(rd == 0, np.inf, ru / rd)
rsi = np.where(rd == 0, 100.0,
np.where(ru == 0, 0.0,
100 - np.round(100 / (1 + rs))))
return rsi
# ══════════════════════════════════════════════════════════════════
# INFORMATIVE — 1D timeframe for trend bias
# ══════════════════════════════════════════════════════════════════
@informative("1d")
def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Daily timeframe indicators:
- Pi Cycle High/Low MAs
- JMA RSI slow (length=420) for macro momentum
- EMA alignment for trend
"""
c = dataframe["close"]
# ── Pi Cycle MAs ─────────────────────────────────────────────
dataframe["pi_low_long"] = c.rolling(471).mean() * (745 / 1000)
dataframe["pi_low_short"] = c.ewm(span=150, adjust=False).mean()
dataframe["pi_hi_long"] = c.rolling(350).mean() * 2.0
dataframe["pi_hi_short"] = c.rolling(111).mean()
# Pi Cycle crossunder signals
dataframe["pi_cycle_low"] = (
(dataframe["pi_low_short"].shift(1) >= dataframe["pi_low_long"].shift(1)) &
(dataframe["pi_low_short"] < dataframe["pi_low_long"])
).astype(int)
dataframe["pi_cycle_hi"] = (
(dataframe["pi_hi_long"].shift(1) >= dataframe["pi_hi_short"].shift(1)) &
(dataframe["pi_hi_long"] < dataframe["pi_hi_short"])
).astype(int)
# ── JMA RSI slow (420) — macro momentum ──────────────────────
dataframe["jrsi2"] = self._jma_mod_rsi(c.values, 420, 2.0)
# ── EMA stack for trend alignment ────────────────────────────
dataframe["ema10"] = c.ewm(span=10, adjust=False).mean()
dataframe["ema20"] = c.ewm(span=20, adjust=False).mean()
dataframe["ema50"] = c.ewm(span=50, adjust=False).mean()
dataframe["ema100"] = c.ewm(span=100, adjust=False).mean()
dataframe["ma200"] = c.rolling(200).mean()
dataframe["ema_bullish"] = (
(dataframe["ema10"] > dataframe["ema20"]) &
(dataframe["ema20"] > dataframe["ema50"]) &
(dataframe["ema50"] > dataframe["ema100"]) &
(dataframe["ema100"] > dataframe["ma200"])
).astype(int)
dataframe["ema_bearish"] = (
(dataframe["ema10"] < dataframe["ema20"]) &
(dataframe["ema20"] < dataframe["ema50"]) &
(dataframe["ema50"] < dataframe["ema100"]) &
(dataframe["ema100"] < dataframe["ma200"])
).astype(int)
return dataframe
# ══════════════════════════════════════════════════════════════════
# MAIN INDICATORS — 4H timeframe
# ══════════════════════════════════════════════════════════════════
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
c = dataframe["close"]
h = dataframe["high"]
lo = dataframe["low"]
# ── 1. JMA Modified RSI (all 3 lengths) ──────────────────────
dataframe["jrsi0"] = self._jma_mod_rsi(c.values, 10, 2.0) # fast
dataframe["jrsi1"] = self._jma_mod_rsi(c.values, 60, 2.0) # mid
# jrsi2 comes from 1d informative
# ── 2. TTM Squeeze ───────────────────────────────────────────
length = 20
# Bollinger Bands
bb_mid = c.rolling(length).mean()
bb_upper = bb_mid + 2.0 * c.rolling(length).std()
# Keltner Channels
tr = pd.concat([
h - lo,
(h - c.shift()).abs(),
(lo - c.shift()).abs()
], axis=1).max(axis=1)
kc_mid = c.ewm(span=length, adjust=False).mean()
kc_upper = kc_mid + 1.0 * tr.ewm(span=length, adjust=False).mean()
# Squeeze ON = BB inside KC
dataframe["squeeze_on"] = (bb_upper < kc_upper).astype(int)
dataframe["squeeze_off"] = (bb_upper >= kc_upper).astype(int)
# Momentum histogram
hh = h.rolling(length).max()
ll = lo.rolling(length).min()
mid = (hh + ll) / 2
delta = c - (mid + kc_mid) / 2
dataframe["ttm_mom"] = delta.rolling(length).apply(
lambda x: np.polyval(np.polyfit(np.arange(len(x)), x, 1), len(x) - 1),
raw=True
)
dataframe["ttm_rising"] = (
dataframe["ttm_mom"] > dataframe["ttm_mom"].shift(1)
).astype(int)
dataframe["ttm_pos"] = (dataframe["ttm_mom"] >= 0).astype(int)
# ── 3. Standard RSI + Divergence ─────────────────────────────
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
# Pivot-based divergence (simplified for Freqtrade compatibility)
rsi = dataframe["rsi"]
lbL = 5
lbR = 5
rLo = 5
rHi = 60
bull_div = pd.Series(0, index=dataframe.index)
bear_div = pd.Series(0, index=dataframe.index)
rsi_arr = rsi.values
low_arr = lo.values
high_arr = h.values
n = len(dataframe)
for i in range(lbL + lbR, n):
# Pivot low check
window = rsi_arr[i - lbL - lbR: i + 1]
if len(window) > 0 and rsi_arr[i - lbR] == np.nanmin(window):
for j in range(i - lbR - rLo, max(i - lbR - rHi, lbL) - 1, -1):
if j < 0:
break
w2 = rsi_arr[j - lbL: j + lbR + 1] if j >= lbL and j + lbR < n else None
if w2 is not None and rsi_arr[j] == np.nanmin(w2):
if rsi_arr[i - lbR] > rsi_arr[j] and low_arr[i - lbR] < low_arr[j]:
bull_div.iloc[i] = 1
break
# Pivot high check
if rsi_arr[i - lbR] == np.nanmax(rsi_arr[i - lbL - lbR: i + 1]):
for j in range(i - lbR - rLo, max(i - lbR - rHi, lbL) - 1, -1):
if j < 0:
break
w2 = rsi_arr[j - lbL: j + lbR + 1] if j >= lbL and j + lbR < n else None
if w2 is not None and rsi_arr[j] == np.nanmax(w2):
if rsi_arr[i - lbR] < rsi_arr[j] and high_arr[i - lbR] > high_arr[j]:
bear_div.iloc[i] = 1
break
dataframe["bull_div"] = bull_div
dataframe["bear_div"] = bear_div
# ── 4. Kalit MA Stack + RSI(2) ───────────────────────────────
dataframe["ma5"] = c.rolling(5).mean()
dataframe["ma200_4h"] = c.rolling(200).mean()
dataframe["rsi2"] = ta.RSI(dataframe, timeperiod=2)
dataframe["kalit_long"] = (
(c > dataframe["ma200_4h"]) &
(c < dataframe["ma5"]) &
(dataframe["rsi2"] < 10)
).astype(int)
dataframe["kalit_short"] = (
(c < dataframe["ma200_4h"]) &
(c > dataframe["ma5"]) &
(dataframe["rsi2"] > 90)
).astype(int)
# ── 5. Pi Cycle position score ───────────────────────────────
# (pulled from 1d informative, used in signal scoring)
# Available as dataframe["1d_pi_cycle_low"] etc after merge
return dataframe
# ══════════════════════════════════════════════════════════════════
# BUY SIGNAL
# ══════════════════════════════════════════════════════════════════
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Multi-indicator buy signal.
Requires confluence of indicators across timeframes.
"""
conditions = []
# ── MUST HAVE: Not in a squeeze (or squeeze just fired) ───────
squeeze_ok = (
(dataframe["squeeze_off"] == 1) |
(dataframe["squeeze_on"].shift(1) == 1) & (dataframe["squeeze_off"] == 1)
)
conditions.append(squeeze_ok)
# ── MUST HAVE: TTM momentum bullish ──────────────────────────
conditions.append(dataframe["ttm_rising"] == 1)
conditions.append(dataframe["ttm_pos"] == 1)
# ── MUST HAVE: JMA RSI mid not overbought ────────────────────
conditions.append(dataframe["jrsi1"] < 75)
conditions.append(dataframe["jrsi1"] > 20)
# ── SHOULD HAVE: At least one of these ───────────────────────
extra = (
(dataframe["bull_div"] == 1) | # RSI bull divergence
(dataframe["kalit_long"] == 1) | # Kalit RSI(2) setup
(dataframe["jrsi0"] < 30) | # JMA RSI fast oversold
(dataframe.get("1d_ema_bullish", 0) == 1) # Daily EMAs bullish
)
conditions.append(extra)
# ── VOLUME: Not dead market ───────────────────────────────────
conditions.append(dataframe["volume"] > 0)
# Combine all conditions
dataframe.loc[
pd.concat([c.to_frame() if isinstance(c, pd.Series) else pd.Series(c, index=dataframe.index).to_frame()
for c in conditions], axis=1).all(axis=1),
"enter_long"
] = 1
return dataframe
# ══════════════════════════════════════════════════════════════════
# SELL SIGNAL
# ══════════════════════════════════════════════════════════════════
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Exit when momentum reverses or overbought.
ROI and stoploss handle most exits — this catches trend reversals.
"""
conditions = []
# TTM momentum turning negative
conditions.append(
(dataframe["ttm_rising"] == 0) &
(dataframe["ttm_pos"] == 0)
)
# OR: JMA RSI fast overbought
conditions.append(dataframe["jrsi0"] > 80)
# OR: RSI bear divergence
conditions.append(dataframe["bear_div"] == 1)
# OR: Pi Cycle HIGH signal (major top warning)
pi_hi_col = "1d_pi_cycle_hi"
if pi_hi_col in dataframe.columns:
conditions.append(dataframe[pi_hi_col] == 1)
dataframe.loc[
pd.concat([c.to_frame() if isinstance(c, pd.Series)
else pd.Series(c, index=dataframe.index).to_frame()
for c in conditions], axis=1).any(axis=1),
"exit_long"
] = 1
return dataframe