#!/usr/bin/env python3
"""
Inside Bar Indicator
====================
Detects Inside Bar patterns and generates trading signals with
Entry, Stop Loss, and Take Profit lines displayed on the chart.
"""
import requests
from typing import List, Dict, Any
# =============================================================================
# CONFIGURATION
# =============================================================================
API_BASE_URL = "https://api.innova-trading.com"
API_KEY = "your_api_key_here" # Get from Dashboard Settings
SYMBOL = "EURUSD"
TIMEFRAME = 60 # H1 (1 hour in minutes)
BARS_TO_FETCH = 500
# Signal parameters
SIGNAL_DELAY_BARS = 3 # Place signal 3 bars after inside bar
SL_MULTIPLIER = 2 # SL = 2x Inside Bar range
TP_RR_RATIOS = [1, 2, 3] # Risk:Reward for TP1, TP2, TP3
LINE_EXTEND_BARS = 10 # Lines extend 10 bars forward
# Colors
COLOR_BUY = "#3b82f6" # Blue
COLOR_SELL = "#ef4444" # Red
COLOR_SL = "#ef4444" # Red
COLOR_TP1 = "#22c55e" # Green
COLOR_TP2 = "#10b981" # Emerald
COLOR_TP3 = "#059669" # Dark green
# =============================================================================
# API FUNCTIONS
# =============================================================================
def get_bars(symbol: str, timeframe: int, limit: int = 500) -> List[Dict]:
"""Fetch OHLC bars from the API"""
response = requests.get(
f"{API_BASE_URL}/api/external/bars",
params={"symbol": symbol, "timeframe": timeframe, "limit": limit},
headers={"Authorization": f"Bearer {API_KEY}"}
)
if response.status_code != 200:
raise Exception(f"Failed to fetch bars: {response.status_code}")
return response.json()["bars"]
def submit_indicator(
indicator_id: str,
symbol: str,
timeframe: int,
indicator_name: str,
points: List[Dict],
lines: List[Dict]
) -> Dict:
"""Submit indicator data to the API"""
response = requests.post(
f"{API_BASE_URL}/api/external/indicators/{indicator_id}",
json={
"symbol": symbol,
"timeframe": timeframe,
"indicator_name": indicator_name,
"version": "1.0",
"points": points,
"lines": lines
},
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
)
if response.status_code not in [200, 201]:
raise Exception(f"Failed to submit: {response.status_code}")
return response.json()
# =============================================================================
# INSIDE BAR DETECTION
# =============================================================================
def detect_inside_bars(bars: List[Dict]) -> List[Dict]:
"""
Detect Inside Bar patterns.
Inside Bar = High < Previous High AND Low > Previous Low
"""
inside_bars = []
for i in range(1, len(bars)):
current = bars[i]
previous = bars[i - 1]
is_inside = (
current["high"] < previous["high"] and
current["low"] > previous["low"]
)
if is_inside:
# Bullish if close > open
is_bullish = current["close"] > current["open"]
inside_bars.append({
"bar_index": i,
"time": current["time"],
"high": current["high"],
"low": current["low"],
"close": current["close"],
"is_bullish": is_bullish,
"range": current["high"] - current["low"]
})
return inside_bars
def generate_signals(bars: List[Dict], inside_bars: List[Dict]) -> tuple:
"""Generate trading signals from Inside Bar patterns"""
points = []
lines = []
signal_count = 0
tf_seconds = TIMEFRAME * 60
for ib in inside_bars:
# Signal bar = 3 bars after inside bar
signal_bar_index = ib["bar_index"] + SIGNAL_DELAY_BARS
if signal_bar_index >= len(bars):
continue
signal_bar = bars[signal_bar_index]
signal_time = signal_bar["time"]
# Calculate risk
ib_range = ib["range"]
risk = ib_range * SL_MULTIPLIER
signal_count += 1
signal_id = f"ib_{signal_count:03d}"
if ib["is_bullish"]:
# BUY signal
entry = signal_bar["close"]
sl = entry - risk
tps = [entry + (risk * rr) for rr in TP_RR_RATIOS]
points.append({
"time": signal_time,
"type": "low",
"price": entry,
"label": "BUY",
"color": COLOR_BUY,
"shape": "arrowUp",
"size": 2
})
else:
# SELL signal
entry = signal_bar["close"]
sl = entry + risk
tps = [entry - (risk * rr) for rr in TP_RR_RATIOS]
points.append({
"time": signal_time,
"type": "high",
"price": entry,
"label": "SELL",
"color": COLOR_SELL,
"shape": "arrowDown",
"size": 2
})
# Add Stop Loss line
lines.append({
"id": f"{signal_id}_sl",
"price": sl,
"start_time": signal_time,
"bars": LINE_EXTEND_BARS,
"label": "SL",
"color": COLOR_SL,
"style": "dashed",
"width": 1
})
# Add Take Profit lines
tp_colors = [COLOR_TP1, COLOR_TP2, COLOR_TP3]
for idx, (tp, color) in enumerate(zip(tps, tp_colors), 1):
lines.append({
"id": f"{signal_id}_tp{idx}",
"price": tp,
"start_time": signal_time,
"bars": LINE_EXTEND_BARS,
"label": f"TP{idx}",
"color": color,
"style": "dotted",
"width": 1
})
return points, lines
# =============================================================================
# MAIN
# =============================================================================
def main():
print(f"Fetching {BARS_TO_FETCH} bars of {SYMBOL} {TIMEFRAME}m...")
bars = get_bars(SYMBOL, TIMEFRAME, BARS_TO_FETCH)
print(f"Received {len(bars)} bars")
print("Detecting Inside Bars...")
inside_bars = detect_inside_bars(bars)
print(f"Found {len(inside_bars)} Inside Bars")
print("Generating signals...")
points, lines = generate_signals(bars, inside_bars)
print(f"Generated {len(points)} signals with {len(lines)} lines")
print("Submitting to API...")
result = submit_indicator(
indicator_id="inside_bar_signals",
symbol=SYMBOL,
timeframe=TIMEFRAME,
indicator_name="Inside Bar Signals",
points=points,
lines=lines
)
print(f"Success! Points: {result['points_received']}, Lines: {result['lines_received']}")
print(f"View at: https://innova-trading.com/trading?symbol={SYMBOL}")
if __name__ == "__main__":
main()
