rules/assets/bollinger_chart/chart.py
"""
Bollinger Bands Chart — with squeeze detection and %B subplot
"""
import os
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
BB_PERIOD = 20
BB_STD = 2.0
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
upper, middle, lower = ta.bbands(close, BB_PERIOD, BB_STD)
bb_pct = ta.bb_percent(close, BB_PERIOD, BB_STD)
bb_width = ta.bb_width(close, BB_PERIOD, BB_STD)
x_labels = df.index.strftime("%Y-%m-%d")
fig = make_subplots(
rows=3, cols=1, shared_xaxes=True,
row_heights=[0.55, 0.22, 0.23], vertical_spacing=0.03,
subplot_titles=[
f"{SYMBOL} Price + Bollinger Bands({BB_PERIOD}, {BB_STD})",
f"Bollinger %B",
f"Bollinger Width (Squeeze Detection)",
],
)
# Candlestick + Bands
fig.add_trace(go.Scatter(
x=x_labels, y=upper, mode="lines", name="Upper BB",
line=dict(color="rgba(100,149,237,0.6)", width=1),
), row=1, col=1)
fig.add_trace(go.Scatter(
x=x_labels, y=lower, mode="lines", name="Lower BB",
line=dict(color="rgba(100,149,237,0.6)", width=1),
fill="tonexty", fillcolor="rgba(100,149,237,0.08)",
), row=1, col=1)
fig.add_trace(go.Scatter(
x=x_labels, y=middle, mode="lines", name=f"SMA({BB_PERIOD})",
line=dict(color="cornflowerblue", width=1, dash="dash"),
), row=1, col=1)
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=df["high"], low=df["low"], close=close,
name="Price",
), row=1, col=1)
# %B
fig.add_trace(go.Scatter(
x=x_labels, y=bb_pct, mode="lines",
name="%B", line=dict(color="yellow", width=1.5),
), row=2, col=1)
fig.add_hline(y=1.0, line_dash="dash", line_color="red", opacity=0.5, row=2, col=1)
fig.add_hline(y=0.0, line_dash="dash", line_color="green", opacity=0.5, row=2, col=1)
fig.add_hline(y=0.5, line_dash="dot", line_color="gray", opacity=0.3, row=2, col=1)
# Width (squeeze)
fig.add_trace(go.Scatter(
x=x_labels, y=bb_width, mode="lines",
name="BB Width", line=dict(color="cyan", width=1.5),
), row=3, col=1)
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} — Bollinger Bands Analysis",
xaxis_rangeslider_visible=False,
height=900,
)
for r in range(1, 4):
fig.update_xaxes(type="category", row=r, col=1)
fig.update_yaxes(side="right", row=r, col=1)
fig.write_html(script_dir / f"{SYMBOL}_bollinger_chart.html")
fig.show()
# Explanation
current_pct = bb_pct.iloc[-1]
current_width = bb_width.iloc[-1]
min_width = pd.Series(bb_width).rolling(20).min().iloc[-1]
if current_pct > 1:
position = "ABOVE upper band — overbought / breakout"
elif current_pct < 0:
position = "BELOW lower band — oversold / breakdown"
elif current_pct > 0.8:
position = "Near upper band — approaching resistance"
elif current_pct < 0.2:
position = "Near lower band — approaching support"
else:
position = "Inside bands — range-bound"
squeeze = "YES — volatility squeeze" if current_width <= min_width * 1.05 else "No"
print(f"\n{SYMBOL} — Bollinger Bands({BB_PERIOD}, {BB_STD}) Analysis")
print(f"Upper: {upper.iloc[-1]:.2f} | Middle: {middle.iloc[-1]:.2f} | Lower: {lower.iloc[-1]:.2f}")
print(f"%B: {current_pct:.4f}")
print(f"Position: {position}")
print(f"Band Width: {current_width:.4f}")
print(f"Squeeze: {squeeze}")
rules/assets/custom_indicator/template.py
"""
Custom Indicator Template — Z-Score Example
Demonstrates the pattern for building custom indicators on top of openalgo's
Rust-core primitives (openalgo.ta) with vectorized NumPy.
No JIT, no warmup: openalgo 2.x computes ta primitives in a compiled Rust core,
so the first call already runs at full speed.
Usage:
from zscore_indicator import zscore
result = zscore(close_prices, period=20)
"""
import numpy as np
import pandas as pd
from openalgo import ta
# =============================================================================
# Core Computation — vectorized NumPy on Rust-core primitives
# =============================================================================
def _compute_zscore(arr: np.ndarray, period: int) -> np.ndarray:
"""
Z-Score: (value - rolling_mean) / rolling_stdev
Measures how many standard deviations the current value is from the mean.
- Z > 2: Extremely high (potential overbought)
- Z > 1: Above average
- Z ~ 0: At average
- Z < -1: Below average
- Z < -2: Extremely low (potential oversold)
Complexity: O(n) — ta.sma and ta.stdev run in the Rust core.
"""
n = len(arr)
result = np.full(n, np.nan)
if period < 2 or n < period:
return result
mean = ta.sma(arr, period)
std = ta.stdev(arr, period)
valid = ~np.isnan(mean) & ~np.isnan(std)
nonzero = valid & (std > 0)
result[nonzero] = (arr[nonzero] - mean[nonzero]) / std[nonzero]
result[valid & (std == 0)] = 0.0
return result
def _compute_zscore_bands(arr: np.ndarray, period: int,
upper_threshold: float,
lower_threshold: float):
"""
Z-Score with upper/lower price bands for signal generation.
Returns: (zscore, upper_band, lower_band, mean_line)
"""
n = len(arr)
zscore_vals = np.full(n, np.nan)
upper_band = np.full(n, np.nan)
lower_band = np.full(n, np.nan)
mean_line = np.full(n, np.nan)
if period < 2 or n < period:
return zscore_vals, upper_band, lower_band, mean_line
mean = ta.sma(arr, period)
std = ta.stdev(arr, period)
valid = ~np.isnan(mean) & ~np.isnan(std)
mean_line[valid] = mean[valid]
upper_band[valid] = mean[valid] + upper_threshold * std[valid]
lower_band[valid] = mean[valid] + lower_threshold * std[valid]
nonzero = valid & (std > 0)
zscore_vals[nonzero] = (arr[nonzero] - mean[nonzero]) / std[nonzero]
zscore_vals[valid & (std == 0)] = 0.0
return zscore_vals, upper_band, lower_band, mean_line
# =============================================================================
# Public API — Handles pandas/numpy/list input
# =============================================================================
def zscore(data, period=20):
"""
Z-Score Indicator
Measures how many standard deviations the current price is from its
rolling mean. Useful for mean-reversion strategies.
Args:
data: Close prices (numpy array, pandas Series, or list)
period: Lookback period (default: 20)
Returns:
Z-Score values (same type as input)
"""
if isinstance(data, pd.Series):
idx = data.index
result = _compute_zscore(data.values.astype(np.float64), period)
return pd.Series(result, index=idx, name=f"ZScore({period})")
arr = np.asarray(data, dtype=np.float64)
return _compute_zscore(arr, period)
def zscore_bands(data, period=20, upper=2.0, lower=-2.0):
"""
Z-Score with price bands.
Args:
data: Close prices
period: Lookback period (default: 20)
upper: Upper threshold in stdev units (default: 2.0)
lower: Lower threshold in stdev units (default: -2.0)
Returns:
Tuple: (zscore, upper_band, lower_band, mean_line)
"""
if isinstance(data, pd.Series):
idx = data.index
z, ub, lb, ml = _compute_zscore_bands(
data.values.astype(np.float64), period, upper, lower)
return (
pd.Series(z, index=idx, name=f"ZScore({period})"),
pd.Series(ub, index=idx, name="Upper"),
pd.Series(lb, index=idx, name="Lower"),
pd.Series(ml, index=idx, name="Mean"),
)
arr = np.asarray(data, dtype=np.float64)
return _compute_zscore_bands(arr, period, upper, lower)
# =============================================================================
# Benchmark — no warmup needed, first call is full speed
# =============================================================================
if __name__ == "__main__":
import time
print("Z-Score Indicator Benchmark")
print("-" * 40)
for size in [10_000, 100_000, 500_000]:
data = np.random.randn(size).cumsum() + 1000
t0 = time.perf_counter()
_ = zscore(data, 20)
elapsed = (time.perf_counter() - t0) * 1000
print(f" zscore({size:>10,} bars): {elapsed:>8.2f}ms")
t0 = time.perf_counter()
_ = zscore_bands(data, 20)
elapsed = (time.perf_counter() - t0) * 1000
print(f" bands ({size:>10,} bars): {elapsed:>8.2f}ms")
rules/assets/dashboard_basic/app.py
"""
Basic Indicator Dashboard — Single symbol with configurable indicators
Run: python app.py
Open: http://127.0.0.1:8050
"""
import os
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import dash
from dash import dcc, html, Input, Output, callback, State
import dash_bootstrap_components as dbc
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
load_dotenv(find_dotenv(), override=False)
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
def fetch_data(symbol, exchange, interval, days=365):
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
df = client.history(
symbol=symbol, exchange=exchange, interval=interval,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
return df
app = dash.Dash(__name__, external_stylesheets=[dbc.themes.DARKLY])
app.layout = dbc.Container([
dbc.Row([
dbc.Col(html.H3("OpenAlgo Indicator Dashboard"), width=8),
], className="mb-3 mt-3"),
dbc.Row([
dbc.Col([
dbc.Label("Symbol"),
dbc.Input(id="symbol-input", value="SBIN", type="text"),
], width=2),
dbc.Col([
dbc.Label("Exchange"),
dbc.Select(id="exchange-select", value="NSE",
options=[{"label": e, "value": e}
for e in ["NSE", "BSE", "NFO", "NSE_INDEX", "MCX"]]),
], width=2),
dbc.Col([
dbc.Label("Interval"),
dbc.Select(id="interval-select", value="D",
options=[{"label": i, "value": i}
for i in ["1m", "5m", "15m", "30m", "1h", "D"]]),
], width=2),
dbc.Col([
dbc.Label("Overlays"),
dbc.Checklist(
id="overlay-select",
options=[
{"label": " EMA(20)", "value": "ema20"},
{"label": " EMA(50)", "value": "ema50"},
{"label": " Bollinger", "value": "bbands"},
{"label": " Supertrend", "value": "supertrend"},
],
value=["ema20"],
inline=True,
),
], width=3),
dbc.Col([
dbc.Button("Load", id="load-btn", color="primary", className="mt-4"),
], width=1),
], className="mb-3"),
dbc.Row([
dbc.Col([
dbc.Label("Subplots"),
dbc.Checklist(
id="subplot-select",
options=[
{"label": " RSI", "value": "rsi"},
{"label": " MACD", "value": "macd"},
{"label": " Volume", "value": "volume"},
{"label": " Stochastic", "value": "stochastic"},
{"label": " ADX", "value": "adx"},
{"label": " OBV", "value": "obv"},
],
value=["rsi", "volume"],
inline=True,
),
], width=12),
], className="mb-3"),
dbc.Row(id="stats-row", className="mb-3"),
dbc.Row([dbc.Col(dcc.Graph(id="main-chart"), width=12)]),
dcc.Interval(id="refresh-interval", interval=60_000, n_intervals=0),
], fluid=True)
@callback(
Output("main-chart", "figure"),
Output("stats-row", "children"),
Input("load-btn", "n_clicks"),
State("symbol-input", "value"),
State("exchange-select", "value"),
State("interval-select", "value"),
State("overlay-select", "value"),
State("subplot-select", "value"),
prevent_initial_call=False,
)
def update_chart(n_clicks, symbol, exchange, interval, overlays, subplots):
days_map = {"1m": 7, "5m": 30, "15m": 90, "30m": 90, "1h": 180, "D": 365}
days = days_map.get(interval, 365)
df = fetch_data(symbol, exchange, interval, days)
close = df["close"]
high = df["high"]
low = df["low"]
if not overlays:
overlays = []
if not subplots:
subplots = []
n_rows = 1 + len(subplots)
heights = [0.5] + [0.5 / max(len(subplots), 1)] * len(subplots) if subplots else [1.0]
titles = [f"{symbol} ({exchange})"] + [s.upper() for s in subplots]
fig = make_subplots(
rows=n_rows, cols=1, shared_xaxes=True,
row_heights=heights, vertical_spacing=0.03,
subplot_titles=titles,
)
fmt = "%Y-%m-%d" if interval == "D" else "%Y-%m-%d %H:%M"
x_labels = df.index.strftime(fmt)
# Candlestick
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price", showlegend=False,
), row=1, col=1)
# Overlays
if "ema20" in overlays:
fig.add_trace(go.Scatter(x=x_labels, y=ta.ema(close, 20),
name="EMA(20)", line=dict(color="cyan", width=1.5)),
row=1, col=1)
if "ema50" in overlays:
fig.add_trace(go.Scatter(x=x_labels, y=ta.ema(close, 50),
name="EMA(50)", line=dict(color="orange", width=1.5)),
row=1, col=1)
if "bbands" in overlays:
upper, mid, lower = ta.bbands(close, 20, 2.0)
fig.add_trace(go.Scatter(x=x_labels, y=upper, name="BB Upper",
line=dict(color="gray", dash="dash", width=1)), row=1, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=lower, name="BB Lower",
line=dict(color="gray", dash="dash", width=1),
fill="tonexty", fillcolor="rgba(128,128,128,0.08)"),
row=1, col=1)
if "supertrend" in overlays:
st, direction = ta.supertrend(high, low, close, 10, 3.0)
st_up = pd.Series(st, index=df.index).where(pd.Series(direction, index=df.index) == -1)
st_down = pd.Series(st, index=df.index).where(pd.Series(direction, index=df.index) == 1)
fig.add_trace(go.Scatter(x=x_labels, y=st_up, name="ST Up",
line=dict(color="lime", width=2)), row=1, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=st_down, name="ST Down",
line=dict(color="red", width=2)), row=1, col=1)
# Subplots
for i, sp in enumerate(subplots, start=2):
if sp == "rsi":
fig.add_trace(go.Scatter(x=x_labels, y=ta.rsi(close, 14),
name="RSI(14)", line=dict(color="yellow", width=1.5)),
row=i, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", opacity=0.5, row=i, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", opacity=0.5, row=i, col=1)
fig.update_yaxes(range=[0, 100], row=i, col=1)
elif sp == "macd":
m, s, h = ta.macd(close, 12, 26, 9)
fig.add_trace(go.Scatter(x=x_labels, y=m, name="MACD",
line=dict(color="cyan", width=1)), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=s, name="Signal",
line=dict(color="orange", width=1)), row=i, col=1)
colors = ["green" if v >= 0 else "red" for v in h]
fig.add_trace(go.Bar(x=x_labels, y=h, name="Hist",
marker_color=colors, opacity=0.6), row=i, col=1)
elif sp == "volume":
vc = ["green" if c >= o else "red" for c, o in zip(close, df["open"])]
fig.add_trace(go.Bar(x=x_labels, y=df["volume"], name="Volume",
marker_color=vc, opacity=0.5), row=i, col=1)
elif sp == "stochastic":
k, d = ta.stochastic(high, low, close, k_period=14, smooth_k=3, d_period=3)
fig.add_trace(go.Scatter(x=x_labels, y=k, name="%K",
line=dict(color="cyan", width=1)), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=d, name="%D",
line=dict(color="orange", width=1)), row=i, col=1)
fig.add_hline(y=80, line_dash="dash", line_color="red", opacity=0.5, row=i, col=1)
fig.add_hline(y=20, line_dash="dash", line_color="green", opacity=0.5, row=i, col=1)
elif sp == "adx":
dp, dm, adx = ta.adx(high, low, close, 14)
fig.add_trace(go.Scatter(x=x_labels, y=dp, name="+DI",
line=dict(color="lime", width=1)), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=dm, name="-DI",
line=dict(color="red", width=1)), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=adx, name="ADX",
line=dict(color="yellow", width=1.5)), row=i, col=1)
fig.add_hline(y=25, line_dash="dash", line_color="gray", opacity=0.5, row=i, col=1)
elif sp == "obv":
fig.add_trace(go.Scatter(x=x_labels, y=ta.obv(close, df["volume"]),
name="OBV", line=dict(color="cyan", width=1.5)),
row=i, col=1)
fig.update_layout(
template="plotly_dark", height=250 + 220 * n_rows,
xaxis_rangeslider_visible=False,
)
for r in range(1, n_rows + 1):
fig.update_xaxes(type="category", row=r, col=1)
fig.update_yaxes(side="right", row=r, col=1)
# Stats cards
try:
quote = client.quotes(symbol=symbol, exchange=exchange)
d = quote.get("data", {})
ltp = d.get("ltp", close.iloc[-1])
prev = d.get("prev_close", close.iloc[-2] if len(close) > 1 else ltp)
except Exception:
ltp = close.iloc[-1]
prev = close.iloc[-2] if len(close) > 1 else ltp
change = ltp - prev
change_pct = (change / prev * 100) if prev != 0 else 0
rsi_val = ta.rsi(close, 14).iloc[-1] if len(close) > 14 else 0
cards = [
("LTP", f"{ltp:,.2f}", "primary"),
("Change", f"{change:+,.2f} ({change_pct:+.2f}%)", "success" if change >= 0 else "danger"),
("RSI(14)", f"{rsi_val:.1f}", "warning" if rsi_val > 70 or rsi_val < 30 else "info"),
("Volume", f"{df['volume'].iloc[-1]:,.0f}", "secondary"),
("EMA(20)", f"{ta.ema(close, 20).iloc[-1]:,.2f}", "info"),
]
stat_cards = [
dbc.Col(dbc.Card(dbc.CardBody([
html.P(label, className="text-muted mb-0", style={"fontSize": "0.75rem"}),
html.H5(value, className="mb-0"),
]), color=color, outline=True), width=2)
for label, value, color in cards
]
return fig, stat_cards
if __name__ == "__main__":
app.run(debug=True, port=8050)
rules/assets/dashboard_multi/app.py
"""
Multi-Timeframe Dashboard — Same symbol across 4 timeframes
Run: python app.py
Open: http://127.0.0.1:8050
"""
import os
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import dash
from dash import dcc, html, Input, Output, State, callback
import dash_bootstrap_components as dbc
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
load_dotenv(find_dotenv(), override=False)
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
TIMEFRAMES = {
"5m": {"interval": "5m", "days": 7, "fmt": "%H:%M"},
"15m": {"interval": "15m", "days": 30, "fmt": "%m-%d %H:%M"},
"1h": {"interval": "1h", "days": 90, "fmt": "%m-%d %H:%M"},
"D": {"interval": "D", "days": 365, "fmt": "%Y-%m-%d"},
}
def fetch_tf_data(symbol, exchange, interval, days):
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
df = client.history(
symbol=symbol, exchange=exchange, interval=interval,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
return df
app = dash.Dash(__name__, external_stylesheets=[dbc.themes.DARKLY])
app.layout = dbc.Container([
dbc.Row([
dbc.Col(html.H3("Multi-Timeframe Analysis"), width=6),
dbc.Col([
dbc.InputGroup([
dbc.Input(id="symbol-input", value="SBIN", placeholder="Symbol"),
dbc.Select(id="exchange-select", value="NSE",
options=[{"label": e, "value": e}
for e in ["NSE", "BSE", "NFO", "NSE_INDEX"]]),
dbc.Button("Load", id="load-btn", color="primary"),
]),
], width=6),
], className="mb-3 mt-3"),
dbc.Row(id="confluence-row", className="mb-3"),
dbc.Row([dbc.Col(dcc.Graph(id="mtf-chart"), width=12)]),
], fluid=True)
@callback(
Output("mtf-chart", "figure"),
Output("confluence-row", "children"),
Input("load-btn", "n_clicks"),
State("symbol-input", "value"),
State("exchange-select", "value"),
prevent_initial_call=False,
)
def update_mtf(n_clicks, symbol, exchange):
fig = make_subplots(
rows=2, cols=2,
subplot_titles=[f"{symbol} — {tf}" for tf in TIMEFRAMES],
vertical_spacing=0.08, horizontal_spacing=0.05,
)
trends = {}
for idx, (tf_name, tf_cfg) in enumerate(TIMEFRAMES.items()):
row = idx // 2 + 1
col = idx % 2 + 1
try:
df = fetch_tf_data(symbol, exchange, tf_cfg["interval"], tf_cfg["days"])
except Exception:
continue
close = df["close"]
x = df.index.strftime(tf_cfg["fmt"])
fig.add_trace(go.Candlestick(
x=x, open=df["open"], high=df["high"], low=df["low"], close=close,
name=tf_name, showlegend=False,
), row=row, col=col)
ema_20 = ta.ema(close, 20)
ema_50 = ta.ema(close, min(50, len(close) - 1)) if len(close) > 50 else ema_20
fig.add_trace(go.Scatter(x=x, y=ema_20, mode="lines",
name=f"EMA20 {tf_name}", line=dict(color="cyan", width=1),
showlegend=False), row=row, col=col)
fig.add_trace(go.Scatter(x=x, y=ema_50, mode="lines",
name=f"EMA50 {tf_name}", line=dict(color="orange", width=1),
showlegend=False), row=row, col=col)
fig.update_xaxes(type="category", row=row, col=col)
fig.update_xaxes(rangeslider_visible=False, row=row, col=col)
fig.update_yaxes(side="right", row=row, col=col)
# Determine trend
if len(close) > 20:
rsi_val = ta.rsi(close, 14).iloc[-1] if len(close) > 14 else 50
ema_trend = "bullish" if ema_20.iloc[-1] > ema_50.iloc[-1] else "bearish"
trends[tf_name] = {"trend": ema_trend, "rsi": rsi_val}
fig.update_layout(
template="plotly_dark", height=800,
title=f"{symbol} Multi-Timeframe Analysis",
showlegend=False,
)
# Confluence cards
bull_count = sum(1 for v in trends.values() if v["trend"] == "bullish")
total = len(trends)
if bull_count == total and total > 0:
confluence = "STRONG BULLISH — All timeframes aligned"
color = "success"
elif bull_count == 0 and total > 0:
confluence = "STRONG BEARISH — All timeframes aligned"
color = "danger"
else:
confluence = f"MIXED — {bull_count}/{total} bullish"
color = "warning"
cards = [dbc.Col(dbc.Alert(confluence, color=color), width=4)]
for tf, data in trends.items():
c = "success" if data["trend"] == "bullish" else "danger"
cards.append(dbc.Col(dbc.Card(dbc.CardBody([
html.P(tf, className="text-muted mb-0"),
html.H6(f"{data['trend'].upper()} | RSI: {data['rsi']:.1f}"),
]), color=c, outline=True), width=2))
return fig, cards
if __name__ == "__main__":
app.run(debug=True, port=8050)
rules/assets/ema_chart/chart.py
"""
EMA Chart — Exponential Moving Average overlay on candlestick
"""
import os
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd
import plotly.graph_objects as go
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
EMA_FAST = 10
EMA_SLOW = 20
EMA_LONG = 50
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
# Compute EMAs
ema_fast = ta.ema(close, EMA_FAST)
ema_slow = ta.ema(close, EMA_SLOW)
ema_long = ta.ema(close, EMA_LONG)
# Crossover signals
buy_signals = ta.crossover(ema_fast, ema_slow)
sell_signals = ta.crossunder(ema_fast, ema_slow)
x_labels = df.index.strftime("%Y-%m-%d")
fig = go.Figure()
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=df["high"], low=df["low"], close=close,
name="Price",
))
fig.add_trace(go.Scatter(
x=x_labels, y=ema_fast, mode="lines",
name=f"EMA({EMA_FAST})", line=dict(color="cyan", width=1.5),
))
fig.add_trace(go.Scatter(
x=x_labels, y=ema_slow, mode="lines",
name=f"EMA({EMA_SLOW})", line=dict(color="orange", width=1.5),
))
fig.add_trace(go.Scatter(
x=x_labels, y=ema_long, mode="lines",
name=f"EMA({EMA_LONG})", line=dict(color="magenta", width=1, dash="dash"),
))
# Buy signals
buy_idx = [x_labels[i] for i in range(len(buy_signals)) if buy_signals[i]]
buy_prices = [close.iloc[i] for i in range(len(buy_signals)) if buy_signals[i]]
fig.add_trace(go.Scatter(
x=buy_idx, y=buy_prices, mode="markers", name="Buy",
marker=dict(symbol="triangle-up", size=12, color="lime"),
))
# Sell signals
sell_idx = [x_labels[i] for i in range(len(sell_signals)) if sell_signals[i]]
sell_prices = [close.iloc[i] for i in range(len(sell_signals)) if sell_signals[i]]
fig.add_trace(go.Scatter(
x=sell_idx, y=sell_prices, mode="markers", name="Sell",
marker=dict(symbol="triangle-down", size=12, color="red"),
))
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} — EMA({EMA_FAST}/{EMA_SLOW}/{EMA_LONG})",
xaxis_rangeslider_visible=False,
xaxis_type="category",
height=600,
)
fig.update_yaxes(side="right")
fig.write_html(script_dir / f"{SYMBOL}_ema_chart.html")
fig.show()
# Explanation
current_fast = ema_fast.iloc[-1]
current_slow = ema_slow.iloc[-1]
current_price = close.iloc[-1]
trend = "bullish" if current_fast > current_slow else "bearish"
above_long = "above" if current_price > ema_long.iloc[-1] else "below"
print(f"\n{SYMBOL} — EMA Analysis")
print(f"Price: {current_price:.2f}")
print(f"EMA({EMA_FAST}): {current_fast:.2f}")
print(f"EMA({EMA_SLOW}): {current_slow:.2f}")
print(f"EMA({EMA_LONG}): {ema_long.iloc[-1]:.2f}")
print(f"Short-term trend: {trend} (EMA {EMA_FAST} {'>' if trend == 'bullish' else '<'} EMA {EMA_SLOW})")
print(f"Long-term position: Price {above_long} EMA({EMA_LONG})")
rules/assets/live_feed/template.py
"""
Real-Time Indicator Feed — WebSocket streaming with live indicator computation
"""
import os
import time
import numpy as np
from datetime import datetime, timedelta
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
EMA_PERIOD = 20
RSI_PERIOD = 14
BUFFER_SIZE = 200
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
verbose=1,
)
# Pre-fetch historical data for buffer initialization
print(f"Fetching historical data for {SYMBOL} buffer...")
try:
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval="1m",
start_date=(datetime.now() - timedelta(days=2)).strftime("%Y-%m-%d"),
end_date=datetime.now().strftime("%Y-%m-%d"),
)
close_buffer = list(df["close"].values[-BUFFER_SIZE:])
print(f"Buffer initialized with {len(close_buffer)} historical bars")
except Exception as e:
print(f"Could not fetch history: {e}")
close_buffer = []
instruments = [{"exchange": EXCHANGE, "symbol": SYMBOL}]
tick_count = 0
start_time = time.time()
def on_data(data):
global tick_count
ltp = data["data"].get("ltp")
if ltp is None:
return
tick_count += 1
close_buffer.append(float(ltp))
# Keep buffer size capped
if len(close_buffer) > BUFFER_SIZE:
close_buffer.pop(0)
if len(close_buffer) >= max(EMA_PERIOD, RSI_PERIOD + 1):
arr = np.array(close_buffer, dtype=np.float64)
ema_val = ta.ema(arr, EMA_PERIOD)[-1]
rsi_val = ta.rsi(arr, RSI_PERIOD)[-1]
# Determine bias
if rsi_val > 70:
bias = "OVERBOUGHT"
elif rsi_val < 30:
bias = "OVERSOLD"
elif ltp > ema_val:
bias = "BULLISH"
else:
bias = "BEARISH"
timestamp = datetime.now().strftime("%H:%M:%S")
print(f"[{timestamp}] {SYMBOL:>10} "
f"LTP:{ltp:>10.2f} | "
f"EMA({EMA_PERIOD}):{ema_val:>10.2f} | "
f"RSI({RSI_PERIOD}):{rsi_val:>6.2f} | "
f"{bias}")
# Connect and subscribe
print(f"\nConnecting to WebSocket...")
client.connect()
client.subscribe_ltp(instruments, on_data_received=on_data)
print(f"Streaming {SYMBOL} on {EXCHANGE}")
print(f"Computing: EMA({EMA_PERIOD}), RSI({RSI_PERIOD})")
print(f"Press Ctrl+C to stop\n")
print(f"{'Time':<12} {'Symbol':>10} {'LTP':>12} {'EMA':>12} {'RSI':>8} {'Bias'}")
print("-" * 70)
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
elapsed = time.time() - start_time
print(f"\n{'='*50}")
print(f"Session Summary")
print(f"Duration: {elapsed:.0f} seconds")
print(f"Ticks received: {tick_count}")
print(f"Ticks/sec: {tick_count / max(elapsed, 1):.1f}")
print(f"Buffer size: {len(close_buffer)} bars")
# Cleanup
client.unsubscribe_ltp(instruments)
client.disconnect()
print("Disconnected.")
rules/assets/macd_chart/chart.py
"""
MACD Chart — Moving Average Convergence Divergence with histogram
"""
import os
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
FAST = 12
SLOW = 26
SIGNAL = 9
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
macd_line, signal_line, histogram = ta.macd(close, FAST, SLOW, SIGNAL)
x_labels = df.index.strftime("%Y-%m-%d")
fig = make_subplots(
rows=2, cols=1, shared_xaxes=True,
row_heights=[0.6, 0.4], vertical_spacing=0.03,
subplot_titles=[f"{SYMBOL} Price", f"MACD({FAST},{SLOW},{SIGNAL})"],
)
# Price
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=df["high"], low=df["low"], close=close,
name="Price", showlegend=False,
), row=1, col=1)
# MACD line
fig.add_trace(go.Scatter(
x=x_labels, y=macd_line, mode="lines",
name="MACD", line=dict(color="cyan", width=1.5),
), row=2, col=1)
# Signal line
fig.add_trace(go.Scatter(
x=x_labels, y=signal_line, mode="lines",
name="Signal", line=dict(color="orange", width=1.5),
), row=2, col=1)
# Histogram
colors = ["rgba(0,200,0,0.6)" if v >= 0 else "rgba(200,0,0,0.6)" for v in histogram]
fig.add_trace(go.Bar(
x=x_labels, y=histogram, name="Histogram",
marker_color=colors,
), row=2, col=1)
# Zero line
fig.add_hline(y=0, line_dash="solid", line_color="gray", opacity=0.3, row=2, col=1)
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} — MACD({FAST},{SLOW},{SIGNAL})",
xaxis_rangeslider_visible=False,
height=700,
)
fig.update_xaxes(type="category", row=1, col=1)
fig.update_xaxes(type="category", row=2, col=1)
fig.update_yaxes(side="right")
fig.write_html(script_dir / f"{SYMBOL}_macd_chart.html")
fig.show()
# Explanation
current_macd = macd_line.iloc[-1]
current_signal = signal_line.iloc[-1]
current_hist = histogram.iloc[-1]
crossover = "bullish" if current_macd > current_signal else "bearish"
momentum = "increasing" if current_hist > histogram.iloc[-2] else "decreasing"
print(f"\n{SYMBOL} — MACD Analysis")
print(f"MACD Line: {current_macd:.4f}")
print(f"Signal Line: {current_signal:.4f}")
print(f"Histogram: {current_hist:.4f}")
print(f"Crossover: {crossover} (MACD {'>' if crossover == 'bullish' else '<'} Signal)")
print(f"Momentum: {momentum}")
rules/assets/multi_indicator/chart.py
"""
Multi-Indicator Chart — Candlestick + EMA + RSI + MACD + Volume
Complete technical analysis in one view
"""
import os
from datetime import datetime, timedelta
from pathlib import Path
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
high = df["high"]
low = df["low"]
volume = df["volume"]
# Compute all indicators
ema_20 = ta.ema(close, 20)
ema_50 = ta.ema(close, 50)
rsi = ta.rsi(close, 14)
macd_line, signal_line, histogram = ta.macd(close, 12, 26, 9)
x_labels = df.index.strftime("%Y-%m-%d")
fig = make_subplots(
rows=4, cols=1, shared_xaxes=True,
row_heights=[0.4, 0.2, 0.2, 0.2],
vertical_spacing=0.025,
subplot_titles=[
f"{SYMBOL} Price + EMA(20/50)",
"RSI(14)",
"MACD(12,26,9)",
"Volume",
],
)
# Row 1: Candlestick + EMA
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price", showlegend=False,
), row=1, col=1)
fig.add_trace(go.Scatter(
x=x_labels, y=ema_20, mode="lines",
name="EMA(20)", line=dict(color="cyan", width=1.5),
), row=1, col=1)
fig.add_trace(go.Scatter(
x=x_labels, y=ema_50, mode="lines",
name="EMA(50)", line=dict(color="orange", width=1.5),
), row=1, col=1)
# Row 2: RSI
fig.add_trace(go.Scatter(
x=x_labels, y=rsi, mode="lines",
name="RSI(14)", line=dict(color="yellow", width=1.5),
), row=2, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", opacity=0.5, row=2, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", opacity=0.5, row=2, col=1)
fig.update_yaxes(range=[0, 100], row=2, col=1)
# Row 3: MACD
fig.add_trace(go.Scatter(
x=x_labels, y=macd_line, mode="lines",
name="MACD", line=dict(color="cyan", width=1),
), row=3, col=1)
fig.add_trace(go.Scatter(
x=x_labels, y=signal_line, mode="lines",
name="Signal", line=dict(color="orange", width=1),
), row=3, col=1)
hist_colors = ["rgba(0,200,0,0.6)" if v >= 0 else "rgba(200,0,0,0.6)" for v in histogram]
fig.add_trace(go.Bar(
x=x_labels, y=histogram, name="Histogram",
marker_color=hist_colors,
), row=3, col=1)
fig.add_hline(y=0, line_color="gray", opacity=0.3, row=3, col=1)
# Row 4: Volume
vol_colors = ["green" if c >= o else "red" for c, o in zip(close, df["open"])]
fig.add_trace(go.Bar(
x=x_labels, y=volume, name="Volume",
marker_color=vol_colors, opacity=0.5,
), row=4, col=1)
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} — Complete Technical Analysis",
xaxis_rangeslider_visible=False,
height=1000,
showlegend=True,
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
)
for r in range(1, 5):
fig.update_xaxes(type="category", row=r, col=1)
fig.update_yaxes(side="right", row=r, col=1)
fig.write_html(script_dir / f"{SYMBOL}_multi_indicator_chart.html")
fig.show()
# Summary
print(f"\n{SYMBOL} — Technical Analysis Summary")
print(f"{'='*50}")
print(f"Price: {close.iloc[-1]:.2f}")
print(f"EMA(20): {ema_20.iloc[-1]:.2f} {'(above)' if close.iloc[-1] > ema_20.iloc[-1] else '(below)'}")
print(f"EMA(50): {ema_50.iloc[-1]:.2f} {'(above)' if close.iloc[-1] > ema_50.iloc[-1] else '(below)'}")
print(f"RSI(14): {rsi.iloc[-1]:.2f}")
print(f"MACD: {macd_line.iloc[-1]:.4f} | Signal: {signal_line.iloc[-1]:.4f}")
print(f"Volume: {volume.iloc[-1]:,.0f}")
# Overall bias
bullish_count = 0
if close.iloc[-1] > ema_20.iloc[-1]: bullish_count += 1
if close.iloc[-1] > ema_50.iloc[-1]: bullish_count += 1
if ema_20.iloc[-1] > ema_50.iloc[-1]: bullish_count += 1
if rsi.iloc[-1] > 50: bullish_count += 1
if macd_line.iloc[-1] > signal_line.iloc[-1]: bullish_count += 1
print(f"\nOverall Bias: {bullish_count}/5 bullish conditions met")
if bullish_count >= 4:
print("Assessment: STRONGLY BULLISH")
elif bullish_count >= 3:
print("Assessment: MODERATELY BULLISH")
elif bullish_count == 2:
print("Assessment: NEUTRAL / MIXED")
elif bullish_count == 1:
print("Assessment: MODERATELY BEARISH")
else:
print("Assessment: STRONGLY BEARISH")
rules/assets/rsi_chart/chart.py
"""
RSI Chart — Relative Strength Index with overbought/oversold zones
"""
import os
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
RSI_PERIOD = 14
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
rsi = ta.rsi(close, RSI_PERIOD)
x_labels = df.index.strftime("%Y-%m-%d")
fig = make_subplots(
rows=2, cols=1, shared_xaxes=True,
row_heights=[0.65, 0.35], vertical_spacing=0.03,
subplot_titles=[f"{SYMBOL} Price", f"RSI({RSI_PERIOD})"],
)
# Price
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=df["high"], low=df["low"], close=close,
name="Price", showlegend=False,
), row=1, col=1)
# RSI
fig.add_trace(go.Scatter(
x=x_labels, y=rsi, mode="lines",
name=f"RSI({RSI_PERIOD})", line=dict(color="yellow", width=1.5),
), row=2, col=1)
# Overbought/Oversold zones
fig.add_hline(y=70, line_dash="dash", line_color="red", opacity=0.7, row=2, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", opacity=0.7, row=2, col=1)
fig.add_hline(y=50, line_dash="dot", line_color="gray", opacity=0.3, row=2, col=1)
# Color zones
fig.add_hrect(y0=70, y1=100, fillcolor="red", opacity=0.05, row=2, col=1)
fig.add_hrect(y0=0, y1=30, fillcolor="green", opacity=0.05, row=2, col=1)
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} — RSI({RSI_PERIOD}) Analysis",
xaxis_rangeslider_visible=False,
height=700,
)
fig.update_xaxes(type="category", row=1, col=1)
fig.update_xaxes(type="category", row=2, col=1)
fig.update_yaxes(range=[0, 100], row=2, col=1)
fig.update_yaxes(side="right")
fig.write_html(script_dir / f"{SYMBOL}_rsi_chart.html")
fig.show()
# Explanation
current_rsi = rsi.iloc[-1]
if current_rsi > 70:
zone = "OVERBOUGHT (>70) — Price may be stretched, potential pullback"
elif current_rsi < 30:
zone = "OVERSOLD (<30) — Price may be undervalued, potential bounce"
elif current_rsi > 50:
zone = "Bullish zone (50-70) — Momentum favors buyers"
else:
zone = "Bearish zone (30-50) — Momentum favors sellers"
print(f"\n{SYMBOL} — RSI({RSI_PERIOD}) Analysis")
print(f"Current RSI: {current_rsi:.2f}")
print(f"Zone: {zone}")
rules/assets/scanner/template.py
"""
Multi-Symbol Indicator Scanner — Screens stocks by technical conditions
"""
import os
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SCAN_TYPE = "rsi-oversold" # Change to desired scan
EXCHANGE = "NSE"
INTERVAL = "D"
DAYS = 365
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
# Watchlist: NIFTY 50
NIFTY50 = [
"ADANIENT", "ADANIPORTS", "APOLLOHOSP", "ASIANPAINT", "AXISBANK",
"BAJAJ-AUTO", "BAJFINANCE", "BAJAJFINSV", "BPCL", "BHARTIARTL",
"BRITANNIA", "CIPLA", "COALINDIA", "DIVISLAB", "DRREDDY",
"EICHERMOT", "GRASIM", "HCLTECH", "HDFCBANK", "HDFCLIFE",
"HEROMOTOCO", "HINDALCO", "HINDUNILVR", "ICICIBANK", "INDUSINDBK",
"INFY", "ITC", "JSWSTEEL", "KOTAKBANK", "LT",
"M&M", "MARUTI", "NESTLEIND", "NTPC", "ONGC",
"POWERGRID", "RELIANCE", "SBILIFE", "SBIN", "SUNPHARMA",
"TCS", "TATACONSUM", "TATAMOTORS", "TATASTEEL", "TECHM",
"TITAN", "ULTRACEMCO", "UPL", "WIPRO",
]
def fetch_data(symbol):
end_date = datetime.now().date()
start_date = end_date - timedelta(days=DAYS)
try:
df = client.history(
symbol=symbol, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
return df
except Exception as e:
print(f" Error fetching {symbol}: {e}")
return None
def scan_rsi_oversold(df, symbol):
close = df["close"]
rsi = ta.rsi(close, 14)
val = rsi.iloc[-1]
if val < 30:
return {"symbol": symbol, "ltp": close.iloc[-1], "rsi": round(val, 2), "signal": "OVERSOLD"}
return None
def scan_rsi_overbought(df, symbol):
close = df["close"]
rsi = ta.rsi(close, 14)
val = rsi.iloc[-1]
if val > 70:
return {"symbol": symbol, "ltp": close.iloc[-1], "rsi": round(val, 2), "signal": "OVERBOUGHT"}
return None
def scan_ema_crossover(df, symbol):
close = df["close"]
ema_fast = ta.ema(close, 10)
ema_slow = ta.ema(close, 20)
cross = ta.crossover(ema_fast, ema_slow)
# Check last 3 bars for recent crossover
if any(cross[-3:]):
return {"symbol": symbol, "ltp": close.iloc[-1],
"ema10": round(ema_fast.iloc[-1], 2),
"ema20": round(ema_slow.iloc[-1], 2),
"signal": "EMA CROSS UP"}
return None
def scan_supertrend_buy(df, symbol):
close = df["close"]
high = df["high"]
low = df["low"]
st, direction = ta.supertrend(high, low, close, 10, 3.0)
direction = pd.Series(direction, index=df.index)
# Direction changed to uptrend in last 3 bars
buy = (direction == -1) & (direction.shift(1) == 1)
if any(buy.iloc[-3:]):
return {"symbol": symbol, "ltp": close.iloc[-1],
"supertrend": round(st[-1], 2),
"signal": "SUPERTREND BUY"}
return None
def scan_volume_spike(df, symbol):
vol = df["volume"]
avg_vol = ta.sma(vol, 20)
if vol.iloc[-1] > 2 * avg_vol.iloc[-1]:
return {"symbol": symbol, "ltp": df["close"].iloc[-1],
"volume": int(vol.iloc[-1]),
"avg_volume": int(avg_vol.iloc[-1]),
"ratio": round(vol.iloc[-1] / avg_vol.iloc[-1], 2),
"signal": "VOLUME SPIKE"}
return None
# Scan dispatch
SCANNERS = {
"rsi-oversold": scan_rsi_oversold,
"rsi-overbought": scan_rsi_overbought,
"ema-crossover": scan_ema_crossover,
"supertrend-buy": scan_supertrend_buy,
"volume-spike": scan_volume_spike,
}
scanner_fn = SCANNERS.get(SCAN_TYPE)
if not scanner_fn:
print(f"Unknown scan type: {SCAN_TYPE}")
print(f"Available: {', '.join(SCANNERS.keys())}")
exit(1)
print(f"Scanning {len(NIFTY50)} symbols for: {SCAN_TYPE}")
print(f"Exchange: {EXCHANGE} | Interval: {INTERVAL}")
print("-" * 60)
results = []
for i, symbol in enumerate(NIFTY50, 1):
print(f" [{i}/{len(NIFTY50)}] {symbol}...", end="", flush=True)
df = fetch_data(symbol)
if df is not None and len(df) > 50:
match = scanner_fn(df, symbol)
if match:
results.append(match)
print(f" MATCH")
else:
print(f" -")
else:
print(f" skip")
print(f"\n{'='*60}")
print(f"Scan: {SCAN_TYPE} | Date: {datetime.now().strftime('%Y-%m-%d %H:%M')}")
print(f"Found: {len(results)} / {len(NIFTY50)} symbols")
print(f"{'='*60}")
if results:
df_results = pd.DataFrame(results)
print(df_results.to_string(index=False))
output_file = script_dir / f"{SCAN_TYPE}_results.csv"
df_results.to_csv(output_file, index=False)
print(f"\nResults saved to: {output_file}")
else:
print("No symbols matched the scan criteria.")
rules/assets/streamlit_basic/app.py
"""
Single-Symbol Streamlit Dashboard — Technical Indicator Analysis
Run: streamlit run app.py
Open: http://localhost:8501
"""
import os
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import streamlit as st
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
load_dotenv(find_dotenv(), override=False)
st.set_page_config(page_title="OpenAlgo Indicators", layout="wide")
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
INTERVAL_DAYS = {"1m": 2, "5m": 7, "15m": 30, "1h": 90, "D": 365}
def fetch_data(symbol, exchange, interval):
days = INTERVAL_DAYS.get(interval, 365)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
df = client.history(
symbol=symbol, exchange=exchange, interval=interval,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
return df
# --- Sidebar ---
st.sidebar.title("Settings")
symbol = st.sidebar.text_input("Symbol", value="SBIN")
exchange = st.sidebar.selectbox("Exchange", ["NSE", "BSE", "NFO", "NSE_INDEX"])
interval = st.sidebar.selectbox("Interval", ["1m", "5m", "15m", "1h", "D"], index=4)
overlays = st.sidebar.multiselect(
"Overlay Indicators",
["EMA(20)", "EMA(50)", "Bollinger Bands", "Supertrend"],
default=["EMA(20)"],
)
subplots = st.sidebar.multiselect(
"Subplot Indicators",
["RSI", "MACD", "Volume", "Stochastic", "ADX", "OBV"],
default=["RSI", "Volume"],
)
auto_refresh = st.sidebar.checkbox("Auto-refresh", value=False)
refresh_sec = st.sidebar.slider("Refresh interval (sec)", 5, 60, 30,
disabled=not auto_refresh)
# --- Fetch Data ---
try:
df = fetch_data(symbol, exchange, interval)
except Exception as e:
st.error(f"Error fetching data: {e}")
st.stop()
close = df["close"]
high = df["high"]
low = df["low"]
volume = df["volume"]
fmt = "%H:%M" if interval in ("1m", "5m", "15m") else "%Y-%m-%d"
x_labels = df.index.strftime(fmt)
# --- Stats Metrics ---
st.title(f"{symbol} Technical Analysis")
try:
quote = client.quotes(symbol=symbol, exchange=exchange)
d = quote.get("data", {})
ltp = d.get("ltp", close.iloc[-1])
prev = d.get("prev_close", close.iloc[-2] if len(close) > 1 else ltp)
change = ltp - prev
change_pct = (change / prev) * 100 if prev else 0
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric("LTP", f"{ltp:,.2f}", f"{change:+.2f}")
c2.metric("Change %", f"{change_pct:+.2f}%")
c3.metric("Volume", f"{d.get('volume', int(volume.iloc[-1])):,}")
c4.metric("RSI(14)", f"{ta.rsi(close, 14).iloc[-1]:.1f}")
ema20_val = ta.ema(close, 20).iloc[-1]
c5.metric("EMA(20)", f"{ema20_val:,.2f}",
f"{'Above' if ltp > ema20_val else 'Below'}")
except Exception:
pass
# --- Build Chart ---
n_rows = 1 + len(subplots)
row_heights = [0.5] + [0.5 / len(subplots)] * len(subplots) if subplots else [1.0]
fig = make_subplots(
rows=n_rows, cols=1, shared_xaxes=True,
row_heights=row_heights, vertical_spacing=0.03,
)
# Row 1: Candlestick + overlays
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price", showlegend=False,
), row=1, col=1)
if "EMA(20)" in overlays:
fig.add_trace(go.Scatter(x=x_labels, y=ta.ema(close, 20),
name="EMA(20)", line=dict(color="cyan", width=1)),
row=1, col=1)
if "EMA(50)" in overlays:
fig.add_trace(go.Scatter(x=x_labels, y=ta.ema(close, 50),
name="EMA(50)", line=dict(color="orange", width=1)),
row=1, col=1)
if "Bollinger Bands" in overlays:
upper, mid, lower = ta.bbands(close, 20, 2.0)
fig.add_trace(go.Scatter(x=x_labels, y=upper, name="BB Upper",
line=dict(color="gray", dash="dash")), row=1, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=lower, name="BB Lower",
line=dict(color="gray", dash="dash"),
fill="tonexty", fillcolor="rgba(128,128,128,0.1)"),
row=1, col=1)
if "Supertrend" in overlays:
st_line, st_dir = ta.supertrend(high, low, close, 10, 3.0)
st_dir_s = pd.Series(st_dir, index=df.index)
colors = ["green" if d == -1 else "red" for d in st_dir]
for j in range(len(x_labels)):
if j == 0:
continue
fig.add_trace(go.Scatter(
x=[x_labels[j - 1], x_labels[j]],
y=[st_line[j - 1], st_line[j]],
mode="lines", showlegend=False,
line=dict(color=colors[j], width=2),
), row=1, col=1)
# Dynamic subplots
for i, sp in enumerate(subplots, start=2):
if sp == "RSI":
rsi = ta.rsi(close, 14)
fig.add_trace(go.Scatter(x=x_labels, y=rsi, name="RSI(14)",
line=dict(color="yellow", width=1)), row=i, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", row=i, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=i, col=1)
elif sp == "MACD":
m, s, h = ta.macd(close, 12, 26, 9)
fig.add_trace(go.Scatter(x=x_labels, y=m, name="MACD",
line=dict(color="cyan")), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=s, name="Signal",
line=dict(color="orange")), row=i, col=1)
colors_h = ["green" if v >= 0 else "red" for v in h]
fig.add_trace(go.Bar(x=x_labels, y=h, name="Hist",
marker_color=colors_h, opacity=0.6), row=i, col=1)
elif sp == "Volume":
vc = ["green" if c >= o else "red" for c, o in zip(close, df["open"])]
fig.add_trace(go.Bar(x=x_labels, y=volume, name="Volume",
marker_color=vc, opacity=0.5), row=i, col=1)
elif sp == "Stochastic":
k, d = ta.stochastic(high, low, close, k_period=14, smooth_k=3, d_period=3)
fig.add_trace(go.Scatter(x=x_labels, y=k, name="%K",
line=dict(color="cyan", width=1)), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=d, name="%D",
line=dict(color="orange", width=1)), row=i, col=1)
fig.add_hline(y=80, line_dash="dash", line_color="red", opacity=0.5, row=i, col=1)
fig.add_hline(y=20, line_dash="dash", line_color="green", opacity=0.5, row=i, col=1)
elif sp == "ADX":
plus_di, minus_di, adx_val = ta.adx(high, low, close, 14)
fig.add_trace(go.Scatter(x=x_labels, y=adx_val, name="ADX",
line=dict(color="white", width=1.5)), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=plus_di, name="+DI",
line=dict(color="green", width=1)), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=minus_di, name="-DI",
line=dict(color="red", width=1)), row=i, col=1)
elif sp == "OBV":
fig.add_trace(go.Scatter(x=x_labels, y=ta.obv(close, volume),
name="OBV", line=dict(color="magenta", width=1)),
row=i, col=1)
fig.update_layout(
template="plotly_dark", height=200 + 250 * n_rows,
xaxis_rangeslider_visible=False,
)
for r in range(1, n_rows + 1):
fig.update_xaxes(type="category", row=r, col=1)
fig.update_yaxes(side="right", row=r, col=1)
st.plotly_chart(fig, use_container_width=True)
# --- Auto-refresh ---
if auto_refresh:
import time
time.sleep(refresh_sec)
st.rerun()
rules/assets/streamlit_multi/app.py
"""
Multi-Timeframe Streamlit Dashboard — Same symbol across 4 timeframes
Run: streamlit run app.py
Open: http://localhost:8501
"""
import os
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import streamlit as st
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
load_dotenv(find_dotenv(), override=False)
st.set_page_config(page_title="Multi-Timeframe Analysis", layout="wide")
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
TIMEFRAMES = {
"5m": {"interval": "5m", "days": 7, "fmt": "%H:%M"},
"15m": {"interval": "15m", "days": 30, "fmt": "%m-%d %H:%M"},
"1h": {"interval": "1h", "days": 90, "fmt": "%m-%d %H:%M"},
"D": {"interval": "D", "days": 365, "fmt": "%Y-%m-%d"},
}
def fetch_tf_data(symbol, exchange, interval, days):
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
df = client.history(
symbol=symbol, exchange=exchange, interval=interval,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
return df
# --- Sidebar ---
st.sidebar.title("Settings")
symbol = st.sidebar.text_input("Symbol", value="SBIN")
exchange = st.sidebar.selectbox("Exchange", ["NSE", "BSE", "NFO", "NSE_INDEX"])
auto_refresh = st.sidebar.checkbox("Auto-refresh", value=False)
refresh_sec = st.sidebar.slider("Refresh interval (sec)", 10, 120, 60,
disabled=not auto_refresh)
# --- Title ---
st.title(f"{symbol} Multi-Timeframe Analysis")
# --- Fetch and Chart ---
trends = {}
# 2x2 grid
row1 = st.columns(2)
row2 = st.columns(2)
all_cols = [row1[0], row1[1], row2[0], row2[1]]
for idx, (tf_name, tf_cfg) in enumerate(TIMEFRAMES.items()):
with all_cols[idx]:
st.subheader(f"{tf_name}")
try:
df = fetch_tf_data(symbol, exchange, tf_cfg["interval"], tf_cfg["days"])
except Exception as e:
st.warning(f"Could not load {tf_name}: {e}")
continue
if len(df) < 20:
st.warning(f"Insufficient data for {tf_name}")
continue
close = df["close"]
x = df.index.strftime(tf_cfg["fmt"])
fig = go.Figure()
# Candlestick
fig.add_trace(go.Candlestick(
x=x, open=df["open"], high=df["high"], low=df["low"], close=close,
showlegend=False,
))
# EMA overlays
ema_20 = ta.ema(close, 20)
ema_50 = ta.ema(close, min(50, len(close) - 1)) if len(close) > 50 else ema_20
fig.add_trace(go.Scatter(x=x, y=ema_20, name="EMA(20)",
line=dict(color="cyan", width=1), showlegend=False))
fig.add_trace(go.Scatter(x=x, y=ema_50, name="EMA(50)",
line=dict(color="orange", width=1), showlegend=False))
fig.update_layout(
template="plotly_dark", height=350,
xaxis_rangeslider_visible=False, xaxis_type="category",
margin=dict(l=10, r=10, t=10, b=10),
)
fig.update_yaxes(side="right")
st.plotly_chart(fig, use_container_width=True)
# Trend determination
rsi_val = ta.rsi(close, 14).iloc[-1] if len(close) > 14 else 50
ema_trend = "Bullish" if ema_20.iloc[-1] > ema_50.iloc[-1] else "Bearish"
trends[tf_name] = {"trend": ema_trend, "rsi": rsi_val, "ltp": close.iloc[-1]}
# --- Confluence Summary ---
if trends:
st.markdown("---")
st.subheader("Confluence Summary")
bull_count = sum(1 for v in trends.values() if v["trend"] == "Bullish")
total = len(trends)
if bull_count == total:
st.success(f"STRONG BULLISH -- All {total} timeframes aligned")
elif bull_count == 0:
st.error(f"STRONG BEARISH -- All {total} timeframes aligned")
else:
st.warning(f"MIXED -- {bull_count}/{total} timeframes bullish")
# Trend cards
cols = st.columns(len(trends))
for i, (tf, data) in enumerate(trends.items()):
with cols[i]:
delta_color = "normal" if data["trend"] == "Bullish" else "inverse"
st.metric(
label=tf,
value=data["trend"],
delta=f"RSI: {data['rsi']:.1f}",
delta_color=delta_color,
)
# --- Auto-refresh ---
if auto_refresh:
import time
time.sleep(refresh_sec)
st.rerun()
rules/assets/supertrend_chart/chart.py
"""
Supertrend Chart — Direction-colored trend overlay on candlestick
"""
import os
from datetime import datetime, timedelta
from pathlib import Path
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
ST_PERIOD = 10
ST_MULTIPLIER = 3.0
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
high = df["high"]
low = df["low"]
st, direction = ta.supertrend(high, low, close, ST_PERIOD, ST_MULTIPLIER)
# Split into uptrend and downtrend segments
st_series = pd.Series(st, index=df.index)
direction_series = pd.Series(direction, index=df.index)
st_up = st_series.where(direction_series == -1)
st_down = st_series.where(direction_series == 1)
# Direction change signals
buy_signals = (direction_series == -1) & (direction_series.shift(1) == 1)
sell_signals = (direction_series == 1) & (direction_series.shift(1) == -1)
x_labels = df.index.strftime("%Y-%m-%d")
fig = go.Figure()
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price",
))
fig.add_trace(go.Scatter(
x=x_labels, y=st_up, mode="lines",
name="Supertrend (Up)", line=dict(color="lime", width=2),
))
fig.add_trace(go.Scatter(
x=x_labels, y=st_down, mode="lines",
name="Supertrend (Down)", line=dict(color="red", width=2),
))
# Buy signals
buy_idx = [x_labels[i] for i in range(len(buy_signals)) if buy_signals.iloc[i]]
buy_prices = [low.iloc[i] * 0.99 for i in range(len(buy_signals)) if buy_signals.iloc[i]]
fig.add_trace(go.Scatter(
x=buy_idx, y=buy_prices, mode="markers", name="Buy Signal",
marker=dict(symbol="triangle-up", size=14, color="lime"),
))
# Sell signals
sell_idx = [x_labels[i] for i in range(len(sell_signals)) if sell_signals.iloc[i]]
sell_prices = [high.iloc[i] * 1.01 for i in range(len(sell_signals)) if sell_signals.iloc[i]]
fig.add_trace(go.Scatter(
x=sell_idx, y=sell_prices, mode="markers", name="Sell Signal",
marker=dict(symbol="triangle-down", size=14, color="red"),
))
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} — Supertrend({ST_PERIOD}, {ST_MULTIPLIER})",
xaxis_rangeslider_visible=False,
xaxis_type="category",
height=600,
)
fig.update_yaxes(side="right")
fig.write_html(script_dir / f"{SYMBOL}_supertrend_chart.html")
fig.show()
# Explanation
current_dir = direction_series.iloc[-1]
current_st = st_series.iloc[-1]
trend = "UPTREND (bullish)" if current_dir == -1 else "DOWNTREND (bearish)"
print(f"\n{SYMBOL} — Supertrend({ST_PERIOD}, {ST_MULTIPLIER}) Analysis")
print(f"Current Price: {close.iloc[-1]:.2f}")
print(f"Supertrend Level: {current_st:.2f}")
print(f"Direction: {trend}")
if current_dir == -1:
print(f"Support at: {current_st:.2f} (stop-loss reference)")
else:
print(f"Resistance at: {current_st:.2f} (stop-loss reference)")
rules/custom-indicators.md
# Custom Indicators — Building with NumPy + openalgo Primitives
## Architecture
Since openalgo 2.0, all built-in indicators run in a compiled Rust core. Custom indicators get the same speed by composition:
1. **Core computation**: vectorized NumPy that composes openalgo `ta` primitives (`ta.sma`, `ta.stdev`, `ta.bbands`, ... — all Rust-backed, all O(n))
2. **Public wrapper**: a plain Python function that handles pandas Series / numpy / list inputs and preserves the index
3. **No JIT, no warmup**: there is nothing to compile — the first call runs at full speed
---
## Template: Simple Custom Indicator
```python
import numpy as np
import pandas as pd
from openalgo import ta
def _compute_zscore(arr: np.ndarray, period: int) -> np.ndarray:
"""Z-Score: (value - mean) / stdev over rolling period. Fully vectorized."""
mean = ta.sma(arr, period) # Rust core
std = ta.stdev(arr, period) # Rust core
z = np.full(len(arr), np.nan)
valid = ~np.isnan(mean) & ~np.isnan(std)
nonzero = valid & (std > 0)
z[nonzero] = (arr[nonzero] - mean[nonzero]) / std[nonzero]
z[valid & (std == 0)] = 0.0
return z
def zscore(data, period=20):
"""Z-Score indicator with pandas/numpy support."""
if isinstance(data, pd.Series):
idx = data.index
result = _compute_zscore(data.values.astype(np.float64), period)
return pd.Series(result, index=idx, name=f"ZScore({period})")
return _compute_zscore(np.asarray(data, dtype=np.float64), period)
```
---
## Template: Multi-Output Custom Indicator
Squeeze Momentum — Bollinger Bands inside Keltner Channel means volatility is compressed. With `ta` primitives this is a few lines:
```python
import numpy as np
from openalgo import ta
def squeeze(high, low, close,
bb_period=20, bb_mult=2.0,
kc_period=20, kc_atr=10, kc_mult=1.5):
"""Squeeze Momentum: returns (squeeze_on, momentum).
squeeze_on: boolean array — True while BB is inside KC (volatility compressed)
momentum: distance of close from the midpoint of the recent range
"""
bb_upper, bb_mid, bb_lower = ta.bbands(close, bb_period, bb_mult)
kc_upper, kc_mid, kc_lower = ta.keltner(high, low, close, kc_period, kc_atr, kc_mult)
squeeze_on = (bb_upper < kc_upper) & (bb_lower > kc_lower)
# Momentum: close vs midpoint of (highest high + lowest low + sma) / range midline
hh = ta.highest(high, kc_period)
ll = ta.lowest(low, kc_period)
midline = (hh + ll) / 2.0
momentum = np.asarray(close, dtype=np.float64) - (midline + bb_mid) / 2.0
return squeeze_on, momentum
```
---
## NumPy Rules (MUST FOLLOW)
### DO
- **Compose from `ta` primitives first** — they run in the Rust core and are O(n)
- Use `np.full(n, np.nan)` to initialize output arrays
- Vectorize with array expressions, `np.where`, and boolean masks
- Guard divisions with masks or `np.errstate(invalid="ignore", divide="ignore")`
- Respect the NaN warm-up that primitives emit (mask on `~np.isnan(...)`)
- Return float64 numpy arrays from core functions
### DO NOT
- Never reimplement an indicator that already exists in `openalgo.ta` (100+ available)
- Never write per-bar Python loops over large arrays when a vectorized form exists
- Never divide by a rolling value without masking zeros/NaN
- Never drop the warm-up NaNs silently — downstream signal code must see them
### Path-Dependent Indicators
Some indicators carry sequential state (each bar depends on the previous output). Before writing a loop, check whether a primitive already provides the recursion: `ta.ema` (exponential), `ta.atr` (Wilder), `ta.supertrend` (band-flip logic), `ta.sar`. If the recursion is genuinely custom, a plain Python loop still works — keep it O(n), operate on float64 numpy arrays, and note that it will be slower than vectorized code on very large inputs.
### NaN Handling Pattern
```python
def my_indicator(arr: np.ndarray, period: int) -> np.ndarray:
base = ta.sma(arr, period) # NaN for the first period-1 bars
out = np.full(len(arr), np.nan)
m = ~np.isnan(base) # only compute where inputs are valid
out[m] = arr[m] - base[m]
return out
```
---
## Using openalgo Primitives as Building Blocks
All public `ta` methods accept numpy/pandas/list and run in the Rust core:
```python
from openalgo import ta
# Rolling math: ta.sma, ta.ema, ta.wma, ta.stdev, ta.highest, ta.lowest
# Price action: ta.true_range, ta.atr, ta.change, ta.roc
# Bands/channels: ta.bbands, ta.keltner, ta.donchian
# Signals: ta.crossover, ta.crossunder, ta.exrem, ta.rising, ta.falling
def my_channel(high, low, close, period=20):
"""Custom channel composed entirely from Rust-core primitives."""
upper = ta.highest(high, period)
lower = ta.lowest(low, period)
mid = ta.sma(close, period)
width = np.where(mid != 0, (upper - lower) / mid * 100.0, np.nan)
return upper, mid, lower, width
```
---
## Performance Tips
1. **Compose from primitives**: every `ta` call is Rust — chaining a few primitives beats hand-written Python every time
2. **Pre-compute shared arrays**: if multiple outputs need the same rolling value, compute it once
3. **Vectorize**: one array expression over 500k bars is fast; 500k loop iterations in Python are not
4. **No warmup needed**: benchmark directly — there is no JIT compile on the first call
5. **Test with large arrays**: always benchmark on 100k+ bars to verify O(n) scaling
```python
# Benchmark pattern
import time
import numpy as np
data = np.random.randn(500_000).cumsum() + 1000
t0 = time.perf_counter()
_ = my_indicator(data, 20)
elapsed = (time.perf_counter() - t0) * 1000
print(f"my_indicator(500k bars): {elapsed:.2f}ms")
```
rules/dashboard-patterns.md
# Dashboard Patterns — Plotly Dash Web Applications
## Basic Dash App Structure
```python
import dash
from dash import dcc, html, Input, Output, callback
import dash_bootstrap_components as dbc
app = dash.Dash(__name__, external_stylesheets=[dbc.themes.DARKLY])
app.layout = dbc.Container([
dbc.Row([
dbc.Col(html.H3("OpenAlgo Indicator Dashboard"), width=12),
]),
dbc.Row([
dbc.Col([
dbc.Label("Symbol"),
dbc.Input(id="symbol-input", value="SBIN", type="text"),
], width=3),
dbc.Col([
dbc.Label("Exchange"),
dbc.Select(id="exchange-select", value="NSE",
options=[{"label": e, "value": e}
for e in ["NSE", "BSE", "NFO", "NSE_INDEX"]]),
], width=3),
dbc.Col([
dbc.Label("Interval"),
dbc.Select(id="interval-select", value="D",
options=[{"label": i, "value": i}
for i in ["1m", "5m", "15m", "1h", "D"]]),
], width=3),
dbc.Col([
dbc.Button("Update", id="update-btn", color="primary", className="mt-4"),
], width=3),
], className="mb-3"),
dbc.Row([
dbc.Col(dcc.Graph(id="main-chart"), width=12),
]),
], fluid=True)
@callback(
Output("main-chart", "figure"),
Input("update-btn", "n_clicks"),
Input("symbol-input", "value"),
Input("exchange-select", "value"),
Input("interval-select", "value"),
)
def update_chart(n_clicks, symbol, exchange, interval):
# Fetch data and compute indicators
df = fetch_data(symbol, exchange, interval)
fig = create_chart(df, symbol)
return fig
if __name__ == "__main__":
app.run(debug=True, port=8050)
```
---
## Multi-Indicator Dashboard Layout
```python
app.layout = dbc.Container([
# Header
dbc.Row([
dbc.Col(html.H3("Technical Analysis Dashboard"), width=8),
dbc.Col([
dbc.InputGroup([
dbc.Input(id="symbol-input", value="SBIN", placeholder="Symbol"),
dbc.Button("Load", id="load-btn", color="primary"),
]),
], width=4),
], className="mb-3"),
# Indicator selectors
dbc.Row([
dbc.Col([
dbc.Label("Overlays"),
dbc.Checklist(
id="overlay-select",
options=[
{"label": "EMA(20)", "value": "ema20"},
{"label": "EMA(50)", "value": "ema50"},
{"label": "Bollinger Bands", "value": "bbands"},
{"label": "Supertrend", "value": "supertrend"},
],
value=["ema20"],
inline=True,
),
], width=6),
dbc.Col([
dbc.Label("Subplots"),
dbc.Checklist(
id="subplot-select",
options=[
{"label": "RSI", "value": "rsi"},
{"label": "MACD", "value": "macd"},
{"label": "Volume", "value": "volume"},
{"label": "ADX", "value": "adx"},
{"label": "Stochastic", "value": "stochastic"},
],
value=["rsi", "volume"],
inline=True,
),
], width=6),
], className="mb-3"),
# Charts
dbc.Row([
dbc.Col(dcc.Graph(id="main-chart"), width=12),
]),
# Stats cards
dbc.Row(id="stats-row", className="mt-3"),
# Auto-refresh
dcc.Interval(id="refresh-interval", interval=60_000, n_intervals=0),
], fluid=True)
```
---
## Callback Pattern: Dynamic Subplots
```python
from plotly.subplots import make_subplots
import plotly.graph_objects as go
@callback(
Output("main-chart", "figure"),
Input("load-btn", "n_clicks"),
Input("symbol-input", "value"),
Input("overlay-select", "value"),
Input("subplot-select", "value"),
)
def update_chart(n_clicks, symbol, overlays, subplots):
df = fetch_data(symbol, "NSE", "D")
close = df["close"]
high = df["high"]
low = df["low"]
x_labels = df.index.strftime("%Y-%m-%d")
n_rows = 1 + len(subplots)
row_heights = [0.5] + [0.5 / len(subplots)] * len(subplots) if subplots else [1.0]
fig = make_subplots(
rows=n_rows, cols=1, shared_xaxes=True,
row_heights=row_heights, vertical_spacing=0.03,
)
# Row 1: Candlestick + overlays
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price", showlegend=False,
), row=1, col=1)
if "ema20" in overlays:
fig.add_trace(go.Scatter(x=x_labels, y=ta.ema(close, 20),
name="EMA(20)", line=dict(color="cyan")),
row=1, col=1)
if "bbands" in overlays:
upper, mid, lower = ta.bbands(close, 20, 2.0)
fig.add_trace(go.Scatter(x=x_labels, y=upper, name="BB Upper",
line=dict(color="gray", dash="dash")), row=1, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=lower, name="BB Lower",
line=dict(color="gray", dash="dash"),
fill="tonexty", fillcolor="rgba(128,128,128,0.1)"),
row=1, col=1)
# Dynamic subplots
for i, sp in enumerate(subplots, start=2):
if sp == "rsi":
fig.add_trace(go.Scatter(x=x_labels, y=ta.rsi(close, 14),
name="RSI(14)", line=dict(color="yellow")),
row=i, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", row=i, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=i, col=1)
elif sp == "macd":
m, s, h = ta.macd(close, 12, 26, 9)
fig.add_trace(go.Scatter(x=x_labels, y=m, name="MACD",
line=dict(color="cyan")), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=s, name="Signal",
line=dict(color="orange")), row=i, col=1)
elif sp == "volume":
colors = ["green" if c >= o else "red"
for c, o in zip(close, df["open"])]
fig.add_trace(go.Bar(x=x_labels, y=df["volume"], name="Volume",
marker_color=colors, opacity=0.5), row=i, col=1)
fig.update_layout(
template="plotly_dark", height=200 + 250 * n_rows,
xaxis_rangeslider_visible=False,
)
for r in range(1, n_rows + 1):
fig.update_xaxes(type="category", row=r, col=1)
return fig
```
---
## Stats Cards Pattern
```python
@callback(
Output("stats-row", "children"),
Input("load-btn", "n_clicks"),
Input("symbol-input", "value"),
)
def update_stats(n_clicks, symbol):
quote = client.quotes(symbol=symbol, exchange="NSE")
d = quote.get("data", {})
cards = [
("LTP", f"{d.get('ltp', 0):,.2f}"),
("Change", f"{d.get('ltp', 0) - d.get('prev_close', 0):+,.2f}"),
("Change %", f"{((d.get('ltp', 0) / d.get('prev_close', 1)) - 1) * 100:+.2f}%"),
("Volume", f"{d.get('volume', 0):,}"),
("Day Range", f"{d.get('low', 0):,.2f} - {d.get('high', 0):,.2f}"),
]
return [
dbc.Col(dbc.Card(dbc.CardBody([
html.P(label, className="text-muted mb-0", style={"fontSize": "0.8rem"}),
html.H5(value, className="mb-0"),
])), width=2)
for label, value in cards
]
```
---
## Auto-Refresh with WebSocket Alternative
For real-time dashboards, either use `dcc.Interval` for polling or integrate WebSocket data:
```python
# Polling approach (simpler)
dcc.Interval(id="refresh-interval", interval=5_000) # Every 5 seconds
@callback(Output("ltp-display", "children"), Input("refresh-interval", "n_intervals"))
def refresh_ltp(n):
quote = client.quotes(symbol="SBIN", exchange="NSE")
return f"LTP: {quote['data']['ltp']}"
```
---
## Running the Dashboard
```bash
python dashboards/my_dashboard/app.py
# Opens at http://127.0.0.1:8050
```
---
## Streamlit Alternative
For Streamlit-based dashboards instead of Dash, see [streamlit-patterns.md](streamlit-patterns.md). Streamlit offers simpler setup (no callbacks), built-in `st.metric()`, `st.dataframe()`, and `st.download_button()`, at the cost of less fine-grained layout control.
rules/data-fetching.md
# Data Fetching
## OpenAlgo (Indian Markets)
### Setup
```python
import os
from dotenv import find_dotenv, load_dotenv
from openalgo import api
load_dotenv(find_dotenv(), override=False)
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
```
### Historical Data
```python
from datetime import datetime, timedelta
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol="SBIN",
exchange="NSE", # NSE, BSE, NFO, MCX, NSE_INDEX
interval="D", # D, 1h, 30m, 15m, 10m, 5m, 3m, 1m
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
```
### Data Source: Broker API vs DuckDB
The `history()` method supports a `source` parameter to choose between broker API and local DuckDB/Historify database:
```python
# Default: fetch from broker API (rate-limited ~3 req/s)
df = client.history(
symbol="SBIN", exchange="NSE", interval="D",
start_date="2024-01-01", end_date="2025-01-01",
source="api",
)
# Fetch from OpenAlgo DuckDB/Historify database (no rate limit)
df = client.history(
symbol="SBIN", exchange="NSE", interval="D",
start_date="2024-01-01", end_date="2025-01-01",
source="db",
)
# Custom intervals only available with source="db"
df = client.history(
symbol="SBIN", exchange="NSE", interval="3m",
start_date="2025-01-01", end_date="2025-02-01",
source="db",
)
```
| Source | Description | Rate Limit | Intervals |
|--------|-------------|-----------|-----------|
| `"api"` | Broker API (default) | ~3 req/s | `1m`, `3m`, `5m`, `10m`, `15m`, `30m`, `1h`, `D` |
| `"db"` | DuckDB/Historify local DB | None | All standard + any custom interval (see below) |
#### DuckDB Custom Intervals (source="db" only)
DuckDB stores only `1m` and `D` data physically. All other intervals are computed on-the-fly via SQL aggregation with exchange-aware candle alignment (e.g., NSE candles align to 9:15 AM market open).
**Intraday** (aggregated from 1m data):
| Category | Examples | Format |
|----------|----------|--------|
| Standard minutes | `1m`, `5m`, `15m`, `30m` | `{N}m` |
| Custom minutes | `2m`, `3m`, `4m`, `6m`, `7m`, `10m`, `12m`, `20m`, `25m`, `45m` | `{N}m` |
| Standard hours | `1h` | `{N}h` |
| Custom hours | `2h`, `3h`, `4h`, `6h` | `{N}h` |
**Daily-based** (aggregated from D data):
| Category | Examples | Format |
|----------|----------|--------|
| Daily | `D` | `D` |
| Weekly | `W`, `2W`, `3W` | `{N}W` |
| Monthly | `M`, `2M`, `3M`, `6M` | `{N}M` |
| Quarterly | `Q`, `2Q` | `{N}Q` |
| Yearly | `Y`, `2Y` | `{N}Y` |
**Not supported with source="db"**: seconds intervals (`1s`, `5s`), custom days (`2D`, `3D`)
#### When to Use Each Source
Use `source="db"` for:
- Backtesting and bulk data analysis (no rate limiting)
- Scanner dashboards with many symbols
- Custom interval aggregation (`2m`, `3m`, `4h`, `W`, `M`, `Q`, `Y`)
- Multi-timeframe analysis with non-standard intervals
Use `source="api"` (default) for:
- Real-time or near real-time data
- When DuckDB database is not configured
### Data Normalization (ALWAYS DO THIS)
```python
import pandas as pd
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
high = df["high"]
low = df["low"]
open_ = df["open"]
volume = df["volume"]
```
### Available Intervals
```python
response = client.intervals()
# Returns: {"minutes": ["1m", "3m", "5m", "10m", "15m", "30m"],
# "hours": ["1h"], "days": ["D"], "weeks": [], "months": []}
```
### Real-Time Quotes
```python
# Single symbol
quote = client.quotes(symbol="SBIN", exchange="NSE")
# Returns: {open, high, low, ltp, bid, ask, prev_close, volume}
# Multiple symbols
quotes = client.multiquotes(symbols=[
{"symbol": "SBIN", "exchange": "NSE"},
{"symbol": "RELIANCE", "exchange": "NSE"},
{"symbol": "INFY", "exchange": "NSE"},
])
```
### Market Depth (Level 5)
```python
depth = client.depth(symbol="SBIN", exchange="NSE")
# Returns: {open, high, low, ltp, ltq, prev_close, volume, oi,
# totalbuyqty, totalsellqty,
# asks: [{price, quantity}, ...], # 5 levels
# bids: [{price, quantity}, ...]} # 5 levels
```
### Exchange Codes
| Exchange | Code | Example Symbols |
|----------|------|-----------------|
| NSE Equity | `NSE` | SBIN, RELIANCE, INFY, TCS |
| BSE Equity | `BSE` | SBIN, RELIANCE |
| NSE Index | `NSE_INDEX` | NIFTY, BANKNIFTY, FINNIFTY |
| NSE F&O | `NFO` | NIFTY30DEC25FUT, NIFTY30DEC2526000CE |
| MCX | `MCX` | CRUDEOIL, GOLD, SILVER |
---
## yfinance (US/Global Markets)
### Setup
```python
import yfinance as yf
```
No API key needed.
### Historical Data
```python
df = yf.download("AAPL", start="2024-01-01", end="2025-01-01", auto_adjust=True)
df.columns = df.columns.droplevel(1) if isinstance(df.columns, pd.MultiIndex) else df.columns
df.columns = [c.lower() for c in df.columns]
```
### Common US Symbols
| Symbol | Name |
|--------|------|
| AAPL | Apple |
| MSFT | Microsoft |
| GOOGL | Alphabet |
| AMZN | Amazon |
| SPY | S&P 500 ETF |
| QQQ | Nasdaq 100 ETF |
| ^GSPC | S&P 500 Index |
---
## Market Detection Pattern
```python
INDIAN_EXCHANGES = {"NSE", "BSE", "NFO", "MCX", "NSE_INDEX"}
def fetch_data(symbol, exchange="NSE", interval="D", days=365, source="api"):
if exchange in INDIAN_EXCHANGES:
# Use OpenAlgo
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
df = client.history(
symbol=symbol, exchange=exchange, interval=interval,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
source=source,
)
else:
# Use yfinance
df = yf.download(symbol, period=f"{days}d", auto_adjust=True)
df.columns = df.columns.droplevel(1) if isinstance(df.columns, pd.MultiIndex) else df.columns
df.columns = [c.lower() for c in df.columns]
# Normalize
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
return df
```
---
## Option Chain Data
```python
chain = client.optionchain(
underlying="NIFTY",
exchange="NSE_INDEX",
expiry_date="30DEC25",
strike_count=10 # Optional: number of strikes around ATM
)
# Returns: {underlying, underlying_ltp, atm_strike, expiry_date,
# chain: [{strike, ce: {symbol, label, ltp, bid, ask, ...},
# pe: {symbol, label, ltp, bid, ask, ...}}, ...]}
```
---
## Option Greeks
`client.optiongreeks()` returns delta, gamma, theta, vega, rho plus implied volatility for any option contract. Use it for options-aware indicators and scanners: IV smile/skew charts, delta-neutral filters, theta-decay dashboards, IV percentile analysis.
```python
greeks = client.optiongreeks(
symbol="NIFTY25NOV2526000CE", # OpenAlgo option symbol
exchange="NFO",
interest_rate=0.00, # Risk-free rate (annualized)
underlying_symbol="NIFTY", # Optional: spot reference
underlying_exchange="NSE_INDEX",
)
# Returns: {status, symbol, strike, option_type, option_price, spot_price,
# implied_volatility, days_to_expiry, expiry_date, interest_rate,
# greeks: {delta, gamma, theta, vega, rho}}
iv = greeks["implied_volatility"]
delta = greeks["greeks"]["delta"]
```
### Expiry Dates
```python
response = client.expiry(symbol="NIFTY", exchange="NFO", instrumenttype="options")
# Returns: {status, data: ["10-JUL-25", "17-JUL-25", ...], message}
```
### Pattern: IV Smile Across the Chain
Combine `optionchain()` + `optiongreeks()` to build IV-based analytics:
```python
chain = client.optionchain(
underlying="NIFTY", exchange="NSE_INDEX",
expiry_date="30DEC25", strike_count=10,
)
rows = []
for entry in chain["chain"]:
for side in ("ce", "pe"):
leg = entry.get(side)
if not leg:
continue
g = client.optiongreeks(symbol=leg["symbol"], exchange="NFO")
if g.get("status") == "success":
rows.append({
"strike": entry["strike"],
"type": side.upper(),
"iv": g["implied_volatility"],
"delta": g["greeks"]["delta"],
"theta": g["greeks"]["theta"],
})
iv_df = pd.DataFrame(rows) # plot iv vs strike per side for the IV smile
```
Note: `optiongreeks()` is one HTTP call per contract — for full-chain sweeps, batch only the strikes you need (e.g., `strike_count=10`) and cache results in scanners.
rules/indicator-catalog.md
# Indicator Catalog — Complete Reference
All indicators are accessed via `from openalgo import ta`. Every indicator accepts numpy arrays, pandas Series, or lists. Output type matches input type. Since openalgo 2.0, all kernels are computed by a compiled Rust core shipped inside the wheel — full speed from the first call.
---
## Trend Indicators (20)
### Moving Averages
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| SMA | `ta.sma` | `(data, period)` | Array |
| EMA | `ta.ema` | `(data, period)` | Array (first-value seed) |
| WMA | `ta.wma` | `(data, period)` | Array |
| DEMA | `ta.dema` | `(data, period)` | Array |
| TEMA | `ta.tema` | `(data, period)` | Array |
| HMA | `ta.hma` | `(data, period)` | Array |
| VWMA | `ta.vwma` | `(data, volume, period)` | Array |
| ALMA | `ta.alma` | `(data, period=21, offset=0.85, sigma=6.0)` | Array |
| KAMA | `ta.kama` | `(data, length=14, fast_length=2, slow_length=30)` | Array |
| ZLEMA | `ta.zlema` | `(data, period)` | Array |
| T3 | `ta.t3` | `(data, period=21, v_factor=0.7)` | Array |
| FRAMA | `ta.frama` | `(high, low, period=26)` | Array |
| TRIMA | `ta.trima` | `(data, period)` | Array |
| McGinley | `ta.mcginley` | `(data, period=14)` | Array |
| VIDYA | `ta.vidya` | `(data, period=14)` | Array |
### Advanced Trend
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| Supertrend | `ta.supertrend` | `(high, low, close, period=10, multiplier=3.0)` | Tuple: (supertrend, direction) |
| Ichimoku | `ta.ichimoku` | `(high, low, close, conversion=9, base=26, lagging=52, displacement=26)` | Tuple: (conversion, base, span_a, span_b, lagging) |
| ChandeKrollStop | `ta.chande_kroll_stop` | `(high, low, close, p=10, q=9, x=1)` | Tuple: (stop_long, stop_short) |
| Alligator | `ta.alligator` | `(high, low, close)` | Tuple: (jaw, teeth, lips) |
| MA Envelopes | `ta.ma_envelopes` | `(data, period=20, percent=2.5)` | Tuple: (upper, basis, lower) |
**Supertrend direction**: -1 = uptrend (green), 1 = downtrend (red)
---
## Momentum Indicators (9)
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| RSI | `ta.rsi` | `(close, period=14)` | Array (0-100) |
| MACD | `ta.macd` | `(close, fast=12, slow=26, signal=9)` | Tuple: (macd, signal, histogram) |
| Stochastic | `ta.stochastic` | `(high, low, close, k_period=14, smooth_k=3, d_period=3)` | Tuple: (slow_k, slow_d) |
| CCI | `ta.cci` | `(high, low, close, period=20)` | Array |
| Williams %R | `ta.williams_r` | `(high, low, close, period=14)` | Array (0 to -100) |
| BOP | `ta.bop` | `(open, high, low, close)` | Array (-1 to 1) |
| ElderRay | `ta.elder_ray` | `(high, low, close, period=13)` | Tuple: (bull_power, bear_power) |
| Fisher | `ta.fisher` | `(high, low, period=9)` | Tuple: (fisher, trigger) |
| CRSI | `ta.crsi` | `(close, rsi_period=3, streak_rsi=2, pct_rank=100)` | Array |
---
## Volatility Indicators (16)
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| ATR | `ta.atr` | `(high, low, close, period=14)` | Array |
| BollingerBands | `ta.bbands` | `(close, period=20, std_dev=2.0)` | Tuple: (upper, middle, lower) |
| Keltner | `ta.keltner` | `(high, low, close, ema=20, atr=10, mult=2.0)` | Tuple: (upper, middle, lower) |
| Donchian | `ta.donchian` | `(high, low, period=20)` | Tuple: (upper, middle, lower) |
| Chaikin Vol | `ta.chaikin_volatility` | `(high, low, ema=10, roc=10)` | Array |
| NATR | `ta.natr` | `(high, low, close, period=14)` | Array (% of close) |
| True Range | `ta.true_range` | `(high, low, close)` | Array |
| Mass Index | `ta.massindex` | `(high, low, length=10)` | Array |
| BB %B | `ta.bb_percent` | `(close, period=20, std_dev=2.0)` | Array (0-1) |
| BB Width | `ta.bb_width` | `(close, period=20, std_dev=2.0)` | Array |
| Chandelier Exit | `ta.chandelier_exit` | `(high, low, close, period=22, mult=3.0)` | Tuple: (long, short) |
| Historical Vol | `ta.historical_volatility` | `(close, period=20)` | Array |
| Ulcer Index | `ta.ulcer_index` | `(close, period=14)` | Array |
| STARC | `ta.starc` | `(high, low, close, period=15, mult=1.33)` | Tuple: (upper, lower) |
| RVI | `ta.rvi` | `(close, period=14)` | Array |
| Ultimate Osc | `ta.ultimate_oscillator` | `(high, low, close, p1=7, p2=14, p3=28)` | Array |
---
## Volume Indicators (15)
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| OBV | `ta.obv` | `(close, volume)` | Array |
| OBV Smoothed | `ta.obv_smoothed` | `(close, volume, ma_type="None", ma_length=20, ...)` | Array or Tuple |
| VWAP | `ta.vwap` | `(high, low, close, volume, anchor="Session", ...)` | Array |
| MFI | `ta.mfi` | `(high, low, close, volume, period=14)` | Array (0-100) |
| ADL | `ta.adl` | `(high, low, close, volume)` | Array |
| CMF | `ta.cmf` | `(high, low, close, volume, period=20)` | Array |
| EMV | `ta.emv` | `(high, low, volume, length=14, divisor=10000)` | Array |
| Force Index | `ta.force_index` | `(close, volume, length=13)` | Array |
| NVI | `ta.nvi` | `(close, volume)` | Array |
| PVI | `ta.pvi` | `(close, volume, initial_value=100.0)` | Array |
| Volume Osc | `ta.volosc` | `(volume, short=5, long=10)` | Array (%) |
| VROC | `ta.vroc` | `(volume, period=25)` | Array |
| KVO | `ta.kvo` | `(high, low, close, volume, trig=13, fast=34, slow=55)` | Tuple: (kvo, trigger) |
| PVT | `ta.pvt` | `(close, volume)` | Array |
| RVOL | `ta.rvol` | `(volume, period=20)` | Array (relative volume ratio) |
---
## Oscillators (20+)
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| CMO | `ta.cmo` | `(close, period=14)` | Array |
| TRIX | `ta.trix` | `(close, length=18)` | Array |
| ROC | `ta.roc` | `(close, period=12)` | Array (%) |
| AO | `ta.awesome_oscillator` | `(high, low, fast=5, slow=34)` | Array |
| AC | `ta.accelerator_oscillator` | `(high, low, period=5)` | Array |
| PPO | `ta.ppo` | `(close, fast=12, slow=26, signal=9)` | Tuple: (ppo, signal, histogram) |
| PO | `ta.po` | `(close, fast=10, slow=20, ma_type="SMA")` | Array |
| DPO | `ta.dpo` | `(close, period=21, is_centered=False)` | Array |
| Aroon Osc | `ta.aroon_oscillator` | `(high, low, period=14)` | Array |
| StochRSI | `ta.stoch_rsi` | `(close, period=14, k=3, d=3)` | Tuple: (k, d) |
| CHO | `ta.cho` | `(high, low, close, volume, fast=3, slow=10)` | Array |
| CHOP | `ta.chop` | `(high, low, close, period=14)` | Array |
| KST | `ta.kst` | `(close, ...)` | Tuple: (kst, signal) |
| TSI | `ta.tsi` | `(close, long=25, short=13, signal=13)` | Tuple: (tsi, signal) |
| Vortex | `ta.vortex` | `(high, low, close, period=14)` | Tuple: (vi_plus, vi_minus) |
| Gator Osc | `ta.gator_oscillator` | `(high, low, close)` | Tuple: (upper, lower) |
| STC | `ta.stc` | `(close, fast=23, slow=50, cycle=10)` | Array |
| Coppock | `ta.coppock` | `(close, wma=10, long_roc=14, short_roc=11)` | Array |
| UO | `ta.uo_oscillator` | `(high, low, close, p1=7, p2=14, p3=28)` | Array |
---
## Statistical Indicators (9)
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| LINREG | `ta.linreg` | `(close, period=14)` | Array |
| LR Slope | `ta.lrslope` | `(close, period=100, interval=1)` | Array |
| Correlation | `ta.correlation` | `(data1, data2, period=20)` | Array (-1 to 1) |
| Beta | `ta.beta` | `(asset, market, period=252)` | Array |
| Variance | `ta.variance` | `(close, lookback=20, mode="PR", ...)` | Array or Tuple |
| TSF | `ta.tsf` | `(close, period=14)` | Array |
| Median | `ta.median` | `(data, period=5)` | Array |
| Mode | `ta.mode` | `(data, period=5)` | Array |
| Median Bands | `ta.median_bands` | `(close, period=5, mult=1.0)` | Tuple: (upper, median, lower) |
---
## Hybrid / Advanced Indicators (6+)
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| ADX | `ta.adx` | `(high, low, close, period=14)` | Tuple: (di_plus, di_minus, adx) |
| DMI | `ta.dmi` | `(high, low, close, period=14)` | Tuple: (di_plus, di_minus) |
| Aroon | `ta.aroon` | `(high, low, period=14)` | Tuple: (aroon_up, aroon_down) |
| Pivot Points | `ta.pivot_points` | `(high, low, close)` | Tuple: (pivot, r1, s1, r2, s2, r3, s3) |
| Parabolic SAR | `ta.sar` | `(high, low, acceleration=0.02, maximum=0.2)` | Tuple: (sar, trend) |
| Williams Fractals | `ta.williams_fractals` | `(high, low, period=2)` | Tuple: (up_fractals, down_fractals) |
| RWI | `ta.rwi` | `(high, low, close, period=14)` | Tuple: (rwi_high, rwi_low) |
---
## TA-Lib Compatible Indicators (18)
Added in openalgo 2.0. These match TA-Lib definitions exactly (where openalgo intentionally follows TradingView/Pine conventions elsewhere — EMA/ATR/ADX seeding, etc. — the differences are documented in the library's TALIB_COMPATIBILITY notes).
| Indicator | Method | Signature | Returns |
|-----------|--------|-----------|---------|
| Momentum | `ta.mom` | `(data, period=10)` | Array (data - data[period] ago) |
| ROC Percentage | `ta.rocp` | `(data, period=10)` | Array ((price - prev) / prev) |
| ROC Ratio | `ta.rocr` | `(data, period=10)` | Array (price / prev) |
| ROC Ratio 100 | `ta.rocr100` | `(data, period=10)` | Array (price / prev * 100) |
| APO | `ta.apo` | `(data, fast_period=12, slow_period=26, ma_type="SMA")` | Array (MA(fast) - MA(slow)) |
| MidPoint | `ta.midpoint` | `(data, period=14)` | Array ((highest + lowest) / 2 of source) |
| MidPrice | `ta.midprice` | `(high, low, period=14)` | Array ((highest(high) + lowest(low)) / 2) |
| Average Price | `ta.avgprice` | `(open_prices, high, low, close)` | Array ((O + H + L + C) / 4) |
| Median Price | `ta.medprice` | `(high, low)` | Array ((H + L) / 2) |
| Typical Price | `ta.typprice` | `(high, low, close)` | Array ((H + L + C) / 3) |
| Weighted Close | `ta.wclprice` | `(high, low, close)` | Array ((H + L + 2C) / 4) |
| Plus DM | `ta.plus_dm` | `(high, low, period=14)` | Array (Wilder-summed +DM) |
| Minus DM | `ta.minus_dm` | `(high, low, period=14)` | Array (Wilder-summed -DM) |
| DX | `ta.dx` | `(high, low, close, period=14)` | Array (100 * abs(+DI - -DI) / (+DI + -DI)) |
| ADXR | `ta.adxr` | `(high, low, close, period=14)` | Array ((ADX + ADX[period-1] ago) / 2) |
| Stochastic Fast | `ta.stochf` | `(high, low, close, fastk_period=5, fastd_period=3)` | Tuple: (fastk, fastd) |
| LinReg Angle | `ta.linregangle` | `(data, period=14)` | Array (degrees(atan(slope))) |
| LinReg Intercept | `ta.linregintercept` | `(data, period=14)` | Array |
---
## Utility Functions
| Function | Signature | Returns | Description |
|----------|-----------|---------|-------------|
| `ta.crossover` | `(series1, series2)` | Boolean array | True where series1 crosses above series2 |
| `ta.crossunder` | `(series1, series2)` | Boolean array | True where series1 crosses below series2 |
| `ta.cross` | `(series1, series2)` | Boolean array | True where any cross occurs |
| `ta.highest` | `(data, period)` | Array | Rolling maximum over period (O(n)) |
| `ta.lowest` | `(data, period)` | Array | Rolling minimum over period (O(n)) |
| `ta.change` | `(data, length=1)` | Array | Difference from N bars ago |
| `ta.stdev` | `(data, period)` | Array | Rolling standard deviation |
| `ta.exrem` | `(buy, sell)` | Cleaned series | Remove consecutive duplicate signals |
| `ta.flip` | `(series)` | Flipped series | Toggle on/off states |
| `ta.valuewhen` | `(condition, value)` | Array | Value at condition trigger |
| `ta.rising` | `(data, period)` | Boolean array | True when rising for N bars |
| `ta.falling` | `(data, period)` | Boolean array | True when falling for N bars |
rules/indicator-combinations.md
# Indicator Combinations
## Why Combine Indicators
No single indicator is reliable alone. Combining indicators from different categories (trend + momentum + volume) creates confluence zones where multiple signals agree, reducing false signals.
---
## Category Mixing Rules
| Combination | Purpose | Example |
|-------------|---------|---------|
| Trend + Momentum | Confirm trend with momentum | EMA crossover + RSI filter |
| Trend + Volume | Validate trend with volume | Supertrend + OBV |
| Momentum + Volume | Confirm reversals | RSI oversold + MFI oversold |
| Trend + Volatility | Dynamic levels | EMA + Bollinger Bands |
| Multiple Trend | Strong trend confirmation | EMA + Supertrend + ADX |
**Avoid**: Combining indicators from the same category that measure the same thing (e.g., RSI + Stochastic both measure momentum).
---
## Pattern 1: Trend + Momentum (EMA + RSI)
```python
from openalgo import ta
import pandas as pd
# Trend: EMA direction
ema_20 = ta.ema(close, 20)
trend_bullish = close > ema_20
# Momentum: RSI not overbought
rsi = ta.rsi(close, 14)
rsi_ok = rsi < 70 # Not overbought for buys
# Combined signal
buy_raw = pd.Series(ta.crossover(close, ema_20), index=close.index).fillna(False)
buy_filtered = buy_raw & (pd.Series(rsi, index=close.index) < 70)
sell_raw = pd.Series(ta.crossunder(close, ema_20), index=close.index).fillna(False)
sell_filtered = sell_raw & (pd.Series(rsi, index=close.index) > 30)
entries = ta.exrem(buy_filtered.fillna(False), sell_filtered.fillna(False))
exits = ta.exrem(sell_filtered.fillna(False), buy_filtered.fillna(False))
```
---
## Pattern 2: Supertrend + Volume Confirmation
```python
st, direction = ta.supertrend(high, low, close, 10, 3.0)
direction = pd.Series(direction, index=close.index)
obv = ta.obv(close, volume)
obv_sma = ta.sma(obv, 20)
# Supertrend flips to uptrend AND OBV above its SMA (accumulation)
buy = ((direction == -1) & (direction.shift(1) == 1) &
(pd.Series(obv, index=close.index) > pd.Series(obv_sma, index=close.index)))
sell = ((direction == 1) & (direction.shift(1) == -1))
entries = ta.exrem(buy.fillna(False), sell.fillna(False))
exits = ta.exrem(sell.fillna(False), buy.fillna(False))
```
---
## Pattern 3: Triple Screen (Elder)
Uses 3 timeframes and 3 indicator types:
```python
# Screen 1: Weekly trend (higher timeframe)
weekly_ema = ta.ema(weekly_close, 26)
weekly_trend = weekly_close.iloc[-1] > weekly_ema.iloc[-1]
# Screen 2: Daily momentum (oscillator for entry timing)
daily_rsi = ta.rsi(daily_close, 14)
daily_macd, daily_signal, daily_hist = ta.macd(daily_close, 12, 26, 9)
# Screen 3: Intraday entry (tight stop)
if weekly_trend:
# Weekly bullish -> look for daily RSI pullback
buy = (daily_rsi < 40) & (daily_rsi.shift(1) >= 40) & (daily_hist > daily_hist.shift(1))
else:
sell = (daily_rsi > 60) & (daily_rsi.shift(1) <= 60) & (daily_hist < daily_hist.shift(1))
```
---
## Pattern 4: Bollinger + Keltner Squeeze
```python
upper_bb, mid_bb, lower_bb = ta.bbands(close, 20, 2.0)
upper_kc, mid_kc, lower_kc = ta.keltner(high, low, close, 20, 10, 1.5)
# Squeeze: BB inside KC
squeeze_on = (upper_bb < upper_kc) & (lower_bb > lower_kc)
# Momentum (using MACD histogram as proxy)
_, _, hist = ta.macd(close, 12, 26, 9)
# Buy when squeeze releases + positive momentum
squeeze_release = (~squeeze_on) & (squeeze_on.shift(1))
buy = squeeze_release & (hist > 0)
sell = squeeze_release & (hist < 0)
```
---
## Pattern 5: ADX + DI Crossover (Strong Trend Trading)
```python
di_plus, di_minus, adx_val = ta.adx(high, low, close, 14)
di_plus = pd.Series(di_plus, index=close.index)
di_minus = pd.Series(di_minus, index=close.index)
adx_series = pd.Series(adx_val, index=close.index)
# Strong trend filter
strong_trend = adx_series > 25
# DI crossover with ADX filter
buy = (pd.Series(ta.crossover(di_plus, di_minus), index=close.index).fillna(False) & strong_trend)
sell = (pd.Series(ta.crossunder(di_plus, di_minus), index=close.index).fillna(False) & strong_trend)
entries = ta.exrem(buy.fillna(False), sell.fillna(False))
exits = ta.exrem(sell.fillna(False), buy.fillna(False))
```
---
## Pattern 6: Multi-Indicator Score Card
Assign scores to multiple indicator conditions and trade only when score exceeds a threshold:
```python
import numpy as np
score = np.zeros(len(close))
# +1 for each bullish condition
rsi = ta.rsi(close, 14)
ema_20 = ta.ema(close, 20)
ema_50 = ta.ema(close, 50)
_, _, adx = ta.adx(high, low, close, 14)
macd_line, signal_line, _ = ta.macd(close, 12, 26, 9)
score += (close > ema_20).astype(float) # Price above EMA(20)
score += (ema_20 > ema_50).astype(float) # EMA(20) above EMA(50)
score += (rsi > 50).astype(float) # RSI bullish
score += (rsi < 70).astype(float) # RSI not overbought
score += (macd_line > signal_line).astype(float) # MACD bullish
score += (adx > 20).astype(float) # Trend present
score = pd.Series(score, index=close.index)
# Trade when 5+ conditions are met
buy = (score >= 5) & (score.shift(1) < 5)
sell = (score < 3) & (score.shift(1) >= 3)
entries = ta.exrem(buy.fillna(False), sell.fillna(False))
exits = ta.exrem(sell.fillna(False), buy.fillna(False))
```
---
## Charting Confluence Zones
```python
# Highlight bars where multiple indicators agree
from plotly.subplots import make_subplots
import plotly.graph_objects as go
# Create background shading for high-score zones
high_score = score >= 5
for i in range(len(df)):
if high_score.iloc[i]:
fig.add_vrect(
x0=x_labels[max(0, i-1)], x1=x_labels[i],
fillcolor="rgba(0,255,0,0.05)", line_width=0,
row=1, col=1,
)
```
rules/multi-timeframe.md
# Multi-Timeframe Analysis
## Concept
Multi-timeframe analysis (MTF) computes the same indicator across different timeframes (e.g., 5m, 15m, 1h, D) to identify confluence zones where signals align.
---
## Pattern: Fetch Multiple Timeframes
```python
from datetime import datetime, timedelta
SYMBOL = "SBIN"
EXCHANGE = "NSE"
timeframes = {
"5m": {"interval": "5m", "days": 30},
"15m": {"interval": "15m", "days": 60},
"1h": {"interval": "1h", "days": 180},
"D": {"interval": "D", "days": 365 * 3},
}
data = {}
for tf_name, tf_config in timeframes.items():
end_date = datetime.now().date()
start_date = end_date - timedelta(days=tf_config["days"])
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE,
interval=tf_config["interval"],
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
# Normalize
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
data[tf_name] = df
```
---
## Pattern: Same Indicator Across Timeframes
```python
from openalgo import ta
results = {}
for tf_name, df in data.items():
close = df["close"]
results[tf_name] = {
"rsi": ta.rsi(close, 14),
"ema_20": ta.ema(close, 20),
"ema_50": ta.ema(close, 50),
"trend": "bullish" if ta.ema(close, 20).iloc[-1] > ta.ema(close, 50).iloc[-1] else "bearish",
}
# Print MTF summary
print(f"\n{'Timeframe':<10} {'RSI(14)':<10} {'EMA Trend':<12}")
print("-" * 35)
for tf_name in timeframes:
rsi_val = results[tf_name]["rsi"].iloc[-1]
trend = results[tf_name]["trend"]
print(f"{tf_name:<10} {rsi_val:<10.2f} {trend:<12}")
```
---
## Pattern: MTF Confluence Detection
```python
def check_confluence(results):
"""Check if all timeframes agree on direction."""
trends = [results[tf]["trend"] for tf in results]
all_bullish = all(t == "bullish" for t in trends)
all_bearish = all(t == "bearish" for t in trends)
if all_bullish:
return "STRONG BULLISH - All timeframes aligned"
elif all_bearish:
return "STRONG BEARISH - All timeframes aligned"
else:
bull_count = sum(1 for t in trends if t == "bullish")
return f"MIXED - {bull_count}/{len(trends)} bullish"
print(check_confluence(results))
```
---
## Chart: Multi-Timeframe Grid
```python
from plotly.subplots import make_subplots
import plotly.graph_objects as go
tf_list = list(data.keys())
fig = make_subplots(
rows=2, cols=2,
subplot_titles=[f"{SYMBOL} — {tf}" for tf in tf_list],
vertical_spacing=0.08, horizontal_spacing=0.05,
)
for idx, tf in enumerate(tf_list):
row = idx // 2 + 1
col = idx % 2 + 1
df = data[tf]
x = df.index.strftime("%Y-%m-%d %H:%M") if tf != "D" else df.index.strftime("%Y-%m-%d")
fig.add_trace(go.Candlestick(
x=x, open=df["open"], high=df["high"], low=df["low"], close=df["close"],
name=tf, showlegend=False,
), row=row, col=col)
ema_20 = ta.ema(df["close"], 20)
fig.add_trace(go.Scatter(
x=x, y=ema_20, mode="lines", name=f"EMA(20) {tf}",
line=dict(color="cyan", width=1),
), row=row, col=col)
fig.update_xaxes(type="category", row=row, col=col)
fig.update_layout(
template="plotly_dark", height=800,
title=f"{SYMBOL} Multi-Timeframe Analysis",
showlegend=False,
)
for row in range(1, 3):
for col in range(1, 3):
fig.update_xaxes(rangeslider_visible=False, row=row, col=col)
fig.show()
```
---
## Pattern: Higher Timeframe Filter
Use a higher timeframe indicator as a filter for lower timeframe signals:
```python
# Daily trend filter
daily_ema_50 = ta.ema(data["D"]["close"], 50)
daily_trend_bullish = daily_ema_50.iloc[-1] < data["D"]["close"].iloc[-1]
# 5-minute signals (only take if aligned with daily trend)
close_5m = data["5m"]["close"]
rsi_5m = ta.rsi(close_5m, 14)
if daily_trend_bullish:
# Only take buy signals on 5m RSI oversold
buy_signals = rsi_5m < 30
print("Daily trend: BULLISH — Looking for 5m RSI oversold entries")
else:
# Only take sell signals on 5m RSI overbought
sell_signals = rsi_5m > 70
print("Daily trend: BEARISH — Looking for 5m RSI overbought entries")
```
rules/performance.md
# Performance — The Rust Core (openalgo 2.x)
## How Indicators Are Computed
Since openalgo 2.0, every indicator kernel runs in a **compiled Rust core**
(`openalgo._oaindicators`, built with PyO3) that ships inside the wheel:
- `pip install openalgo` is all you need — indicators are built in, no optional
extra, no separate install step
- Numba and llvmlite are **gone** — they are not dependencies and are never used
- No JIT compilation, no warmup, no compile cache — the first call runs at full speed
- abi3 wheels support **Python 3.12, 3.13, and 3.14** (openalgo 2.x requires Python >= 3.12)
- Dependencies are just: `numpy>=2.0`, `pandas>=2.2`, `httpx`, `websocket-client`
```python
from openalgo import ta
import numpy as np
close = np.random.randn(1_000_000).cumsum() + 1000
ema = ta.ema(close, 20) # first call — already full Rust speed, no warmup
```
## What Changed vs openalgo 1.x (Numba era)
| openalgo 1.x (Numba) | openalgo 2.x (Rust) |
|----------------------|---------------------|
| `pip install openalgo[indicators]` extra | Plain `pip install openalgo` |
| First call triggered JIT compile (100-500ms) | No compilation — first call is full speed |
| `.nbi`/`.nbc` cache files in `__pycache__/` | No cache files to manage or clear |
| Warmup patterns (`_warmup()`, tiny-array pre-calls) | Not needed; `_warmup()` is now a no-op |
| Blocked on new Python/NumPy versions | Python 3.12 / 3.13 / 3.14, NumPy 2.x |
| `@njit` custom-indicator templates | Vectorized NumPy + `ta` primitives (see custom-indicators.md) |
Legacy imports like `from openalgo.numba_shim import jit` still work — the shim
returns the function unchanged — but nothing is compiled, so do not write new
code against it.
## Speed Guarantees
- **Every indicator is O(n)** — rolling sums, Wilder/EMA recursions, and
monotonic-deque extrema are implemented in Rust
- Benchmarked head-to-head against TA-Lib on 924k bars: the
regression/statistics family (`linreg`, `tsf`, `stdev`, `cci`, `macd`, ...)
runs **faster than TA-Lib**; the rest are on par
- Reference: [TA-Lib performance comparison](https://github.com/marketcalls/openalgo-python-library/blob/master/benchmark/TALIB_PERF_COMPARE.md)
and [TA-Lib compatibility notes](https://github.com/marketcalls/openalgo-python-library/blob/master/docs/TALIB_COMPATIBILITY.md)
## NumPy Fallback
If the compiled extension is unavailable (e.g., running from a source checkout
without a built wheel), a pure-NumPy fallback in
`openalgo/indicators/_backend.py` computes the **same values**. Installed
wheels always include the Rust core, so user environments never hit the
fallback. Neither path depends on numba.
## Getting the Most Out of It
1. **Pass numpy arrays or pandas Series** — lists are accepted but get converted
on every call; keep data as float64 arrays in hot paths
2. **Compute once, reuse** — if several charts/signals need `ta.atr(h, l, c, 14)`,
compute it once and pass it around
3. **Compose custom indicators from `ta` primitives** — `ta.sma`, `ta.ema`,
`ta.stdev`, `ta.highest`, `ta.lowest`, `ta.true_range` all run in Rust;
building on them keeps custom code fast (see custom-indicators.md)
4. **Vectorize custom math with NumPy** — avoid per-bar Python loops on large
arrays; use array expressions, `np.where`, and boolean masks
5. **Input/output types match** — pass a pandas Series, get a Series back with
the same index; pass numpy, get numpy
## Benchmark Pattern
No warmup call needed — measure directly:
```python
import time
import numpy as np
from openalgo import ta
for size in [100_000, 500_000, 1_000_000]:
data = np.random.randn(size).cumsum() + 1000
t0 = time.perf_counter()
_ = ta.rsi(data, 14)
elapsed = (time.perf_counter() - t0) * 1000
print(f"rsi({size:>10,} bars): {elapsed:>8.2f}ms")
```
## Algorithm Complexity (for custom code)
When writing your own indicator math, keep it O(n):
| Pattern | Complexity | Approach |
|---------|------------|----------|
| Rolling sum/mean | O(n) | `np.cumsum` difference, or `ta.sma` |
| Rolling stdev | O(n) | cumsum of x and x^2, or `ta.stdev` |
| EMA/Wilder recursion | O(n) | `ta.ema` / built-in primitives (Rust) |
| Rolling max/min | O(n) | `ta.highest` / `ta.lowest` (deque-based, in Rust) |
| Per-bar window slice | O(n x period) | Avoid: `data[i-period:i].max()` in a loop |
rules/plotting.md
# Plotting — Plotly Chart Patterns
## Global Rules
1. **Always** use `template="plotly_dark"` for all charts
2. **Candlestick charts** must use `xaxis_type="category"` to avoid weekend/holiday gaps
3. **X-axis labels**: `df.index.strftime("%Y-%m-%d")` for daily, `"%Y-%m-%d %H:%M"` for intraday
4. **Never use icons/emojis** in chart titles or annotations
5. Use `make_subplots` for multi-panel layouts
---
## Candlestick with Indicator Overlay
```python
import plotly.graph_objects as go
x_labels = df.index.strftime("%Y-%m-%d")
fig = go.Figure()
# Candlestick
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=df["high"], low=df["low"], close=df["close"],
name="Price",
))
# EMA overlay
fig.add_trace(go.Scatter(
x=x_labels, y=ema_20, mode="lines",
name="EMA(20)", line=dict(color="cyan", width=1.5),
))
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} — EMA(20)",
xaxis_rangeslider_visible=False,
xaxis_type="category",
height=600,
)
fig.show()
```
---
## Multi-Panel Chart (Subplots)
Standard layout: Price + Overlays on top, Oscillators below.
```python
from plotly.subplots import make_subplots
fig = make_subplots(
rows=3, cols=1, shared_xaxes=True,
row_heights=[0.5, 0.25, 0.25],
vertical_spacing=0.03,
subplot_titles=["Price + Indicators", "RSI(14)", "MACD"],
)
x_labels = df.index.strftime("%Y-%m-%d")
# Row 1: Candlestick + overlays
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price", showlegend=False,
), row=1, col=1)
fig.add_trace(go.Scatter(
x=x_labels, y=ema_20, mode="lines",
name="EMA(20)", line=dict(color="cyan", width=1.5),
), row=1, col=1)
# Row 2: RSI
fig.add_trace(go.Scatter(
x=x_labels, y=rsi, mode="lines",
name="RSI(14)", line=dict(color="yellow", width=1.5),
), row=2, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", opacity=0.5, row=2, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", opacity=0.5, row=2, col=1)
# Row 3: MACD
fig.add_trace(go.Scatter(
x=x_labels, y=macd_line, mode="lines",
name="MACD", line=dict(color="cyan", width=1),
), row=3, col=1)
fig.add_trace(go.Scatter(
x=x_labels, y=signal_line, mode="lines",
name="Signal", line=dict(color="orange", width=1),
), row=3, col=1)
# MACD histogram with color
colors = ["green" if v >= 0 else "red" for v in histogram]
fig.add_trace(go.Bar(
x=x_labels, y=histogram, name="Histogram",
marker_color=colors, opacity=0.6,
), row=3, col=1)
fig.update_layout(
template="plotly_dark",
title=f"{SYMBOL} Technical Analysis",
xaxis_rangeslider_visible=False,
height=900,
)
# Category x-axis for all rows
for i in range(1, 4):
fig.update_xaxes(type="category", row=i, col=1)
fig.show()
```
---
## Bollinger Bands (Fill Between)
```python
upper, middle, lower = ta.bbands(close, 20, 2.0)
fig = go.Figure()
fig.add_trace(go.Scatter(
x=x_labels, y=upper, mode="lines", name="Upper BB",
line=dict(color="rgba(100,149,237,0.5)", width=1),
))
fig.add_trace(go.Scatter(
x=x_labels, y=lower, mode="lines", name="Lower BB",
line=dict(color="rgba(100,149,237,0.5)", width=1),
fill="tonexty", fillcolor="rgba(100,149,237,0.1)",
))
fig.add_trace(go.Scatter(
x=x_labels, y=middle, mode="lines", name="SMA(20)",
line=dict(color="cornflowerblue", width=1),
))
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price",
))
fig.update_layout(
template="plotly_dark", xaxis_rangeslider_visible=False,
xaxis_type="category", height=600,
)
```
---
## Supertrend (Color-Coded Direction)
```python
st, direction = ta.supertrend(high, low, close, 10, 3.0)
# Split into uptrend and downtrend segments
st_up = pd.Series(st, index=df.index).where(direction == -1)
st_down = pd.Series(st, index=df.index).where(direction == 1)
fig = go.Figure()
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close, name="Price",
))
fig.add_trace(go.Scatter(
x=x_labels, y=st_up, mode="lines", name="Supertrend (Up)",
line=dict(color="lime", width=2),
))
fig.add_trace(go.Scatter(
x=x_labels, y=st_down, mode="lines", name="Supertrend (Down)",
line=dict(color="red", width=2),
))
```
---
## Volume Bar Chart
```python
# Color by price direction
vol_colors = ["green" if c >= o else "red" for c, o in zip(close, df["open"])]
fig.add_trace(go.Bar(
x=x_labels, y=volume, name="Volume",
marker_color=vol_colors, opacity=0.5,
), row=volume_row, col=1)
```
---
## Signal Markers (Buy/Sell Arrows)
```python
# Buy signals
buy_idx = df.index[entries == True]
buy_prices = close[entries == True]
fig.add_trace(go.Scatter(
x=buy_idx.strftime("%Y-%m-%d"), y=buy_prices,
mode="markers", name="Buy",
marker=dict(symbol="triangle-up", size=12, color="lime"),
))
# Sell signals
sell_idx = df.index[exits == True]
sell_prices = close[exits == True]
fig.add_trace(go.Scatter(
x=sell_idx.strftime("%Y-%m-%d"), y=sell_prices,
mode="markers", name="Sell",
marker=dict(symbol="triangle-down", size=12, color="red"),
))
```
---
## Save Chart to HTML
```python
fig.write_html(script_dir / f"{SYMBOL}_chart.html")
```
---
## Common Chart Height Guide
| Layout | Height |
|--------|--------|
| Single panel | 500-600px |
| 2 panels | 700px |
| 3 panels | 900px |
| 4 panels | 1000px |
| Dashboard (full) | 1200px |
rules/signal-generation.md
# Signal Generation
## Core Pattern
Every signal generation follows this pipeline:
1. **Compute indicator(s)** using `openalgo.ta`
2. **Generate raw boolean signals** from conditions
3. **Fill NaN with False** (critical before exrem)
4. **Clean with `ta.exrem()`** to remove duplicate consecutive signals
```python
from openalgo import ta
import pandas as pd
# 1. Compute
ema_fast = ta.ema(close, 10)
ema_slow = ta.ema(close, 20)
# 2. Raw signals
buy_raw = pd.Series(ta.crossover(ema_fast, ema_slow), index=close.index)
sell_raw = pd.Series(ta.crossunder(ema_fast, ema_slow), index=close.index)
# 3. Fill NaN
buy_raw = buy_raw.fillna(False)
sell_raw = sell_raw.fillna(False)
# 4. Clean
entries = ta.exrem(buy_raw, sell_raw)
exits = ta.exrem(sell_raw, buy_raw)
```
---
## Crossover / Crossunder
```python
# Series A crosses above Series B
cross_up = ta.crossover(series_a, series_b)
# Series A crosses below Series B
cross_down = ta.crossunder(series_a, series_b)
# Any cross (either direction)
any_cross = ta.cross(series_a, series_b)
```
These return boolean numpy arrays (or pandas Series if input is Series).
---
## Signal Cleaning with exrem()
`ta.exrem(buy, sell)` removes consecutive duplicate signals. After an entry signal, all subsequent entry signals are removed until an exit signal occurs (and vice versa).
```
Raw: BUY BUY BUY SELL SELL BUY
Cleaned: BUY --- --- SELL ---- BUY
```
**ALWAYS call `.fillna(False)` before exrem** — NaN values break the cleaning logic.
---
## Common Signal Patterns
### EMA Crossover
```python
ema_fast = ta.ema(close, 10)
ema_slow = ta.ema(close, 20)
buy = pd.Series(ta.crossover(ema_fast, ema_slow), index=close.index).fillna(False)
sell = pd.Series(ta.crossunder(ema_fast, ema_slow), index=close.index).fillna(False)
```
### RSI Overbought/Oversold
```python
rsi = ta.rsi(close, 14)
buy = pd.Series(ta.crossover(rsi, pd.Series(np.full(len(rsi), 30.0), index=close.index)),
index=close.index).fillna(False)
sell = pd.Series(ta.crossover(pd.Series(np.full(len(rsi), 70.0), index=close.index), rsi),
index=close.index).fillna(False)
```
Or simpler threshold approach:
```python
rsi = ta.rsi(close, 14)
buy = (rsi < 30) & (pd.Series(rsi).shift(1) >= 30) # RSI crosses below 30
sell = (rsi > 70) & (pd.Series(rsi).shift(1) <= 70) # RSI crosses above 70
buy = buy.fillna(False)
sell = sell.fillna(False)
```
### Supertrend Direction Change
```python
st, direction = ta.supertrend(high, low, close, 10, 3.0)
direction = pd.Series(direction, index=close.index)
buy = (direction == -1) & (direction.shift(1) == 1) # Switch to uptrend
sell = (direction == 1) & (direction.shift(1) == -1) # Switch to downtrend
buy = buy.fillna(False)
sell = sell.fillna(False)
```
### MACD Signal Cross
```python
macd_line, signal_line, histogram = ta.macd(close, 12, 26, 9)
buy = pd.Series(ta.crossover(macd_line, signal_line), index=close.index).fillna(False)
sell = pd.Series(ta.crossunder(macd_line, signal_line), index=close.index).fillna(False)
```
### Bollinger Band Breakout
```python
upper, middle, lower = ta.bbands(close, 20, 2.0)
buy = pd.Series(ta.crossover(close, lower), index=close.index).fillna(False) # Price crosses above lower band
sell = pd.Series(ta.crossunder(close, upper), index=close.index).fillna(False) # Price crosses below upper band
```
### ADX Trend Strength Filter
```python
di_plus, di_minus, adx_val = ta.adx(high, low, close, 14)
# Only generate signals when ADX > 25 (strong trend)
trend_filter = pd.Series(adx_val, index=close.index) > 25
# Combine with directional signals
buy = pd.Series(ta.crossover(di_plus, di_minus), index=close.index).fillna(False) & trend_filter
sell = pd.Series(ta.crossunder(di_plus, di_minus), index=close.index).fillna(False) & trend_filter
```
---
## Condition Helpers
```python
# Rising for N bars
is_rising = ta.rising(close, 3) # True when close has risen for 3 consecutive bars
# Falling for N bars
is_falling = ta.falling(close, 3)
# Highest/Lowest over period
hh = ta.highest(high, 20) # 20-bar highest high
ll = ta.lowest(low, 20) # 20-bar lowest low
# Value when condition is true
entry_price = ta.valuewhen(buy_signal, close)
# Flip toggle
state = ta.flip(buy_signal) # Alternates between True/False on each trigger
```
---
## Signal Counting
```python
import pandas as pd
entries_clean = ta.exrem(buy.fillna(False), sell.fillna(False))
exits_clean = ta.exrem(sell.fillna(False), buy.fillna(False))
n_buys = entries_clean.sum()
n_sells = exits_clean.sum()
print(f"Buy signals: {n_buys}, Sell signals: {n_sells}")
```
rules/streamlit-patterns.md
# Streamlit Patterns — Streamlit Web Applications
Alternative to Plotly Dash for building indicator dashboards. Use Streamlit when the user prefers it or asks for `streamlit` specifically.
---
## Basic Streamlit App Structure
```python
import os
from datetime import datetime, timedelta
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import streamlit as st
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
load_dotenv(find_dotenv(), override=False)
st.set_page_config(page_title="OpenAlgo Indicators", layout="wide")
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
def fetch_data(symbol, exchange, interval, days=365):
end_date = datetime.now().date()
start_date = end_date - timedelta(days=days)
df = client.history(
symbol=symbol, exchange=exchange, interval=interval,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
return df
# --- Sidebar ---
st.sidebar.title("Settings")
symbol = st.sidebar.text_input("Symbol", value="SBIN")
exchange = st.sidebar.selectbox("Exchange", ["NSE", "BSE", "NFO", "NSE_INDEX"])
interval = st.sidebar.selectbox("Interval", ["1m", "5m", "15m", "1h", "D"])
overlays = st.sidebar.multiselect(
"Overlay Indicators",
["EMA(20)", "EMA(50)", "Bollinger Bands", "Supertrend"],
default=["EMA(20)"],
)
subplots = st.sidebar.multiselect(
"Subplot Indicators",
["RSI", "MACD", "Volume", "Stochastic", "ADX"],
default=["RSI", "Volume"],
)
# --- Fetch Data ---
df = fetch_data(symbol, exchange, interval)
close = df["close"]
high = df["high"]
low = df["low"]
x_labels = df.index.strftime("%Y-%m-%d %H:%M" if interval != "D" else "%Y-%m-%d")
# --- Build Chart ---
n_rows = 1 + len(subplots)
row_heights = [0.5] + [0.5 / len(subplots)] * len(subplots) if subplots else [1.0]
fig = make_subplots(
rows=n_rows, cols=1, shared_xaxes=True,
row_heights=row_heights, vertical_spacing=0.03,
)
# Candlestick
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price", showlegend=False,
), row=1, col=1)
# Overlays
if "EMA(20)" in overlays:
fig.add_trace(go.Scatter(x=x_labels, y=ta.ema(close, 20),
name="EMA(20)", line=dict(color="cyan", width=1)),
row=1, col=1)
if "EMA(50)" in overlays:
fig.add_trace(go.Scatter(x=x_labels, y=ta.ema(close, 50),
name="EMA(50)", line=dict(color="orange", width=1)),
row=1, col=1)
# Subplots
for i, sp in enumerate(subplots, start=2):
if sp == "RSI":
rsi = ta.rsi(close, 14)
fig.add_trace(go.Scatter(x=x_labels, y=rsi, name="RSI(14)",
line=dict(color="yellow", width=1)), row=i, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", row=i, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=i, col=1)
elif sp == "MACD":
m, s, h = ta.macd(close, 12, 26, 9)
fig.add_trace(go.Scatter(x=x_labels, y=m, name="MACD",
line=dict(color="cyan")), row=i, col=1)
fig.add_trace(go.Scatter(x=x_labels, y=s, name="Signal",
line=dict(color="orange")), row=i, col=1)
elif sp == "Volume":
vc = ["green" if c >= o else "red" for c, o in zip(close, df["open"])]
fig.add_trace(go.Bar(x=x_labels, y=df["volume"], name="Volume",
marker_color=vc, opacity=0.5), row=i, col=1)
fig.update_layout(
template="plotly_dark", height=200 + 250 * n_rows,
xaxis_rangeslider_visible=False,
)
for r in range(1, n_rows + 1):
fig.update_xaxes(type="category", row=r, col=1)
# --- Display ---
st.title(f"{symbol} Technical Analysis")
st.plotly_chart(fig, use_container_width=True)
```
---
## Metrics Row Pattern
```python
# Stats metrics row
quote = client.quotes(symbol=symbol, exchange=exchange)
d = quote.get("data", {})
ltp = d.get("ltp", 0)
prev_close = d.get("prev_close", 1)
change = ltp - prev_close
change_pct = (change / prev_close) * 100 if prev_close else 0
col1, col2, col3, col4 = st.columns(4)
col1.metric("LTP", f"{ltp:,.2f}", f"{change:+.2f}")
col2.metric("Change %", f"{change_pct:+.2f}%")
col3.metric("Volume", f"{d.get('volume', 0):,}")
col4.metric("Day Range", f"{d.get('low', 0):,.2f} - {d.get('high', 0):,.2f}")
```
---
## Multi-Timeframe Grid Pattern
```python
TIMEFRAMES = {
"5m": {"interval": "5m", "days": 7, "fmt": "%H:%M"},
"15m": {"interval": "15m", "days": 30, "fmt": "%m-%d %H:%M"},
"1h": {"interval": "1h", "days": 90, "fmt": "%m-%d %H:%M"},
"D": {"interval": "D", "days": 365, "fmt": "%Y-%m-%d"},
}
st.title(f"{symbol} Multi-Timeframe Analysis")
# 2x2 grid
row1_cols = st.columns(2)
row2_cols = st.columns(2)
all_cols = [row1_cols[0], row1_cols[1], row2_cols[0], row2_cols[1]]
trends = {}
for idx, (tf_name, tf_cfg) in enumerate(TIMEFRAMES.items()):
with all_cols[idx]:
st.subheader(f"{symbol} - {tf_name}")
try:
df = fetch_data(symbol, exchange, tf_cfg["interval"], tf_cfg["days"])
except Exception:
st.warning(f"Could not load {tf_name}")
continue
close = df["close"]
x = df.index.strftime(tf_cfg["fmt"])
fig = go.Figure()
fig.add_trace(go.Candlestick(
x=x, open=df["open"], high=df["high"], low=df["low"], close=close,
showlegend=False,
))
ema_20 = ta.ema(close, 20)
fig.add_trace(go.Scatter(x=x, y=ema_20, name="EMA(20)",
line=dict(color="cyan", width=1), showlegend=False))
fig.update_layout(
template="plotly_dark", height=350,
xaxis_rangeslider_visible=False, xaxis_type="category",
margin=dict(l=10, r=10, t=10, b=10),
)
st.plotly_chart(fig, use_container_width=True)
# Track trend
if len(close) > 20:
ema_50 = ta.ema(close, min(50, len(close) - 1))
trend = "Bullish" if ema_20.iloc[-1] > ema_50.iloc[-1] else "Bearish"
rsi_val = ta.rsi(close, 14).iloc[-1] if len(close) > 14 else 50
trends[tf_name] = {"trend": trend, "rsi": rsi_val}
# Confluence summary
if trends:
bull_count = sum(1 for v in trends.values() if v["trend"] == "Bullish")
total = len(trends)
if bull_count == total:
st.success(f"STRONG BULLISH -- All {total} timeframes aligned")
elif bull_count == 0:
st.error(f"STRONG BEARISH -- All {total} timeframes aligned")
else:
st.warning(f"MIXED -- {bull_count}/{total} bullish")
cols = st.columns(len(trends))
for i, (tf, data) in enumerate(trends.items()):
with cols[i]:
color = "green" if data["trend"] == "Bullish" else "red"
st.metric(tf, data["trend"], f"RSI: {data['rsi']:.1f}")
```
---
## Scanner Table Pattern
```python
st.title("Indicator Scanner")
scan_type = st.sidebar.selectbox("Scan Type", [
"RSI Oversold", "RSI Overbought", "EMA Crossover",
"Supertrend Buy", "Volume Spike",
])
WATCHLIST = ["SBIN", "RELIANCE", "INFY", "HDFCBANK", "TCS", "ICICIBANK"]
if st.sidebar.button("Run Scan"):
progress = st.progress(0)
results = []
for i, sym in enumerate(WATCHLIST):
progress.progress((i + 1) / len(WATCHLIST))
try:
df = fetch_data(sym, "NSE", "D")
close = df["close"]
rsi_val = ta.rsi(close, 14).iloc[-1]
if scan_type == "RSI Oversold" and rsi_val < 30:
results.append({"Symbol": sym, "LTP": close.iloc[-1],
"RSI": round(rsi_val, 2), "Signal": "OVERSOLD"})
elif scan_type == "RSI Overbought" and rsi_val > 70:
results.append({"Symbol": sym, "LTP": close.iloc[-1],
"RSI": round(rsi_val, 2), "Signal": "OVERBOUGHT"})
except Exception:
pass
progress.empty()
if results:
df_results = pd.DataFrame(results)
st.dataframe(df_results, use_container_width=True)
st.download_button("Download CSV", df_results.to_csv(index=False),
file_name=f"{scan_type.lower().replace(' ', '_')}_results.csv")
else:
st.info("No symbols matched the scan criteria.")
```
---
## Auto-Refresh Pattern
```python
import time
# Sidebar refresh control
auto_refresh = st.sidebar.checkbox("Auto-refresh", value=False)
refresh_interval = st.sidebar.slider("Refresh interval (sec)", 5, 60, 10)
# Place at the end of the app
if auto_refresh:
time.sleep(refresh_interval)
st.rerun()
```
---
## Dark Theme CSS Injection
Streamlit uses its own theming. For a dark look consistent with Plotly charts:
```python
st.markdown("""
<style>
.stApp { background-color: #0e1117; }
.stMetric label { color: #fafafa; }
.stMetric [data-testid="stMetricValue"] { color: #fafafa; }
</style>
""", unsafe_allow_html=True)
```
Or create `.streamlit/config.toml` in the app directory:
```toml
[theme]
base = "dark"
primaryColor = "#00d4ff"
backgroundColor = "#0e1117"
secondaryBackgroundColor = "#1a1a2e"
textColor = "#fafafa"
```
---
## Running Streamlit Apps
```bash
cd dashboards/{dashboard_name}
streamlit run app.py
# Opens at http://localhost:8501 in browser
```
---
## When to Use Streamlit vs Dash
| Feature | Streamlit | Plotly Dash |
|---------|-----------|-------------|
| Setup complexity | Minimal (no callbacks) | Moderate (callback decorators) |
| Reactivity | Automatic rerun on input change | Explicit callbacks |
| Layout control | `st.columns()`, `st.sidebar` | `dbc.Row/Col`, full CSS |
| Charts | `st.plotly_chart()` | `dcc.Graph()` |
| Stats display | `st.metric()` built-in | `dbc.Card()` manual |
| Tables | `st.dataframe()` built-in | `dash_table.DataTable` |
| File download | `st.download_button()` built-in | `dcc.Download` component |
| Progress bars | `st.progress()` built-in | Manual |
| Best for | Quick prototypes, single-page apps | Complex multi-page apps, fine-grained control |
rules/symbol-format.md
# OpenAlgo Symbol Format
## Exchange Codes
| Exchange | Code | Description |
|----------|------|-------------|
| NSE | `NSE` | National Stock Exchange equities |
| BSE | `BSE` | Bombay Stock Exchange equities |
| NFO | `NFO` | NSE Futures and Options |
| BFO | `BFO` | BSE Futures and Options |
| CDS | `CDS` | NSE Currency Derivatives |
| BCD | `BCD` | BSE Currency Derivatives |
| MCX | `MCX` | Multi Commodity Exchange |
| NSE_INDEX | `NSE_INDEX` | NSE Indices |
| BSE_INDEX | `BSE_INDEX` | BSE Indices |
## Equity Symbols
Direct base symbol. Examples:
- `SBIN` (NSE/BSE)
- `RELIANCE` (NSE/BSE)
- `INFY` (NSE/BSE)
- `TATAMOTORS` (NSE/BSE)
## Futures Format
`[BaseSymbol][ExpiryDate]FUT`
Examples:
- `BANKNIFTY24APR24FUT` — Bank Nifty futures
- `NIFTY30DEC25FUT` — Nifty futures
- `USDINR10MAY24FUT` — Currency futures
- `CRUDEOILM20MAY24FUT` — MCX crude oil
## Options Format
`[BaseSymbol][ExpiryDate][StrikePrice][CE/PE]`
Examples:
- `NIFTY28MAR2420800CE` — Nifty 20800 Call
- `VEDL25APR24292.5CE` — VEDL 292.5 Call
- `USDINR19APR2482CE` — Currency option
- `CRUDEOIL17APR246750CE` — MCX option
## Common NSE Index Symbols (NSE_INDEX)
| Symbol | Description |
|--------|-------------|
| NIFTY | Nifty 50 |
| BANKNIFTY | Bank Nifty |
| FINNIFTY | Fin Nifty |
| MIDCPNIFTY | Midcap Nifty |
| NIFTYNXT50 | Nifty Next 50 |
| NIFTYIT | Nifty IT |
| NIFTYPHARMA | Nifty Pharma |
| NIFTYAUTO | Nifty Auto |
| NIFTYMETAL | Nifty Metal |
| NIFTYPVTBANK | Nifty Private Bank |
| NIFTYPSUBANK | Nifty PSU Bank |
| NIFTYREALTY | Nifty Realty |
| NIFTYENERGY | Nifty Energy |
| NIFTYFMCG | Nifty FMCG |
| NIFTYMIDCAP50 | Nifty Midcap 50 |
| NIFTYSMLCAP50 | Nifty Smallcap 50 |
| INDIAVIX | India VIX |
## Common BSE Index Symbols (BSE_INDEX)
| Symbol | Description |
|--------|-------------|
| SENSEX | Sensex 30 |
| BANKEX | Bank Index |
| SENSEX50 | Sensex 50 |
## API Usage
```python
# Equity
df = client.history(symbol="SBIN", exchange="NSE", interval="D", ...)
# Index
df = client.history(symbol="NIFTY", exchange="NSE_INDEX", interval="D", ...)
# Futures
df = client.history(symbol="NIFTY30DEC25FUT", exchange="NFO", interval="5m", ...)
# Quotes
quote = client.quotes(symbol="SBIN", exchange="NSE")
# Symbol lookup
info = client.symbol(symbol="NIFTY30DEC25FUT", exchange="NFO")
# Returns: {symbol, brsymbol, exchange, expiry, lotsize, tick_size, token, ...}
# Option chain
chain = client.optionchain(underlying="NIFTY", exchange="NSE_INDEX", expiry_date="30DEC25")
```
rules/websocket-feeds.md
# WebSocket Feeds — Real-Time Data
## Overview
OpenAlgo provides WebSocket streaming for real-time market data. Three subscription modes:
| Mode | Method | Data Fields |
|------|--------|------------|
| **LTP** | `subscribe_ltp()` | Last traded price only |
| **Quote** | `subscribe_quote()` | OHLC + LTP + volume |
| **Depth** | `subscribe_depth()` | Full L5 order book |
## Connection Setup
```python
from openalgo import api
import os
from dotenv import find_dotenv, load_dotenv
load_dotenv(find_dotenv(), override=False)
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
ws_url=os.getenv("OPENALGO_WS_URL", None), # Optional: custom WS URL
verbose=1, # 0=silent, 1=basic, 2=debug
)
# Connect to WebSocket server
client.connect()
```
## Verbose Levels
| Level | Value | Description |
|-------|-------|-------------|
| Silent | `False` or `0` | Errors only (default) |
| Basic | `True` or `1` | Connection, auth, subscription logs |
| Debug | `2` | All data updates including LTP/Quote/Depth |
---
## LTP Subscription
```python
instruments = [
{"exchange": "NSE", "symbol": "SBIN"},
{"exchange": "NSE", "symbol": "RELIANCE"},
{"exchange": "NSE_INDEX", "symbol": "NIFTY"},
]
def on_ltp(data):
symbol = data["symbol"]
ltp = data["data"].get("ltp")
print(f"{symbol}: {ltp}")
client.subscribe_ltp(instruments, on_data_received=on_ltp)
```
## Quote Subscription
```python
def on_quote(data):
symbol = data["symbol"]
d = data["data"]
print(f"{symbol} O:{d.get('open')} H:{d.get('high')} L:{d.get('low')} LTP:{d.get('ltp')} V:{d.get('volume')}")
client.subscribe_quote(instruments, on_data_received=on_quote)
```
## Depth Subscription
```python
def on_depth(data):
symbol = data["symbol"]
d = data["data"]
best_bid = d.get("bids", [{}])[0].get("price", 0)
best_ask = d.get("asks", [{}])[0].get("price", 0)
spread = best_ask - best_bid
print(f"{symbol} Bid:{best_bid} Ask:{best_ask} Spread:{spread:.2f}")
client.subscribe_depth(instruments, on_data_received=on_depth)
```
## Polling Stored Data
After subscribing, data is stored internally and can be polled:
```python
import time
for _ in range(10):
# Get latest LTP data
ltp_data = client.get_ltp()
for exch, symbols in ltp_data.get("ltp", {}).items():
for sym, data in symbols.items():
print(f" {exch}:{sym} = {data.get('ltp')}")
# Get latest quotes
quote_data = client.get_quotes()
for exch, symbols in quote_data.get("quote", {}).items():
for sym, data in symbols.items():
print(f" {exch}:{sym} LTP={data.get('ltp')} V={data.get('volume')}")
time.sleep(1)
```
## Unsubscribe and Disconnect
```python
# Unsubscribe specific instruments
client.unsubscribe_ltp(instruments)
client.unsubscribe_quote(instruments)
client.unsubscribe_depth(instruments)
# Disconnect WebSocket
client.disconnect()
```
---
## Real-Time Indicator Pattern
Compute indicators on live data by maintaining a rolling buffer:
```python
import numpy as np
from openalgo import ta
import time
# Pre-fetch historical data for initial buffer
df = client.history(symbol="SBIN", exchange="NSE", interval="1m",
start_date="2025-02-28", end_date="2025-02-28")
close_buffer = list(df["close"].values)
instruments = [{"exchange": "NSE", "symbol": "SBIN"}]
def on_live_data(data):
ltp = data["data"].get("ltp")
if ltp is None:
return
close_buffer.append(float(ltp))
# Keep last 200 bars
if len(close_buffer) > 200:
close_buffer.pop(0)
if len(close_buffer) >= 20:
arr = np.array(close_buffer, dtype=np.float64)
ema_val = ta.ema(arr, 20)[-1]
rsi_val = ta.rsi(arr, 14)[-1]
print(f"SBIN LTP:{ltp} EMA(20):{ema_val:.2f} RSI(14):{rsi_val:.2f}")
client.connect()
client.subscribe_ltp(instruments, on_data_received=on_live_data)
# Run for 60 seconds
time.sleep(60)
client.unsubscribe_ltp(instruments)
client.disconnect()
```
---
## WebSocket Log Tags
| Tag | Meaning |
|-----|---------|
| `[WS]` | Connection events |
| `[AUTH]` | Authentication |
| `[SUB]` | Subscription operations |
| `[UNSUB]` | Unsubscription |
| `[LTP]` | LTP updates (verbose=2) |
| `[QUOTE]` | Quote updates (verbose=2) |
| `[DEPTH]` | Depth updates (verbose=2) |
| `[ERROR]` | Errors (always shown) |
SKILL.md
---
name: indicator-expert
description: OpenAlgo indicator expert. Use when user asks about technical indicators, charting, plotting indicators, creating custom indicators, building dashboards, real-time feeds, scanning stocks, indicator combinations, or using openalgo.ta. Also triggers for indicator functions (sma, ema, rsi, macd, supertrend, bollinger, atr, adx, ichimoku, stochastic, obv, vwap, crossover, crossunder, exrem).
user-invocable: false
---
# OpenAlgo Indicator Expert Skill
## Environment
- Python 3.12+ (required by openalgo 2.x) with openalgo, pandas, numpy, plotly, dash, streamlit
- Data sources: OpenAlgo (Indian markets via `client.history()`, `client.quotes()`, `client.depth()`), yfinance (US/Global)
- Real-time: OpenAlgo WebSocket (`client.connect()`, `subscribe_ltp`, `subscribe_quote`, `subscribe_depth`)
- Indicators: **openalgo.ta** (ALWAYS — 100+ indicators computed by a compiled Rust core, full speed from the first call)
- Charts: Plotly with `template="plotly_dark"`
- Dashboards: Plotly Dash with `dash-bootstrap-components` (default) OR Streamlit with `st.plotly_chart()` — use Streamlit only when the user explicitly asks for it
- Custom indicators: vectorized NumPy composed from openalgo.ta primitives (Rust core — no JIT, no warmup)
- API keys loaded from single root `.env` via `python-dotenv` + `find_dotenv()` — never hardcode keys
- Scripts go in appropriate directories (charts/, dashboards/, custom_indicators/, scanners/) created on-demand
- Never use icons/emojis in code or logger output
## Critical Rules
1. **ALWAYS use openalgo.ta** for ALL technical indicators. Never reimplement what already exists in the library.
2. **Data normalization**: Always convert DataFrame index to datetime, sort, and strip timezone after fetching.
3. **Signal cleaning**: Always use `ta.exrem()` after generating raw buy/sell signals. Always `.fillna(False)` before exrem.
4. **Plotly dark theme**: All charts use `template="plotly_dark"` with `xaxis type="category"` for candlesticks.
5. **Custom indicators**: Compose from openalgo.ta primitives (`ta.sma`, `ta.stdev`, `ta.bbands`, ...) plus vectorized NumPy. Never reimplement built-ins; no JIT or warmup is needed.
6. **Input flexibility**: openalgo.ta accepts numpy arrays, pandas Series, or lists. Output matches input type.
7. **WebSocket feeds**: Use `client.connect()`, `client.subscribe_ltp()` / `subscribe_quote()` / `subscribe_depth()` for real-time data.
8. **Environment**: Load `.env` from project root via `find_dotenv()` — never hardcode API keys.
9. **Market detection**: If symbol looks Indian (SBIN, RELIANCE, NIFTY), use OpenAlgo. If US (AAPL, MSFT), use yfinance.
10. **Always explain** chart outputs in plain language so traders understand what the indicator shows.
## Data Source Priority
| Market | Data Source | Method | Example Symbols |
|--------|------------|--------|-----------------|
| India (equity) | OpenAlgo | `client.history()` | SBIN, RELIANCE, INFY |
| India (index) | OpenAlgo | `client.history(exchange="NSE_INDEX")` | NIFTY, BANKNIFTY |
| India (F&O) | OpenAlgo | `client.history(exchange="NFO")` | NIFTY30DEC25FUT |
| US/Global | yfinance | `yf.download()` | AAPL, MSFT, SPY |
## OpenAlgo API Methods for Data
| Method | Purpose | Returns |
|--------|---------|---------|
| `client.history(symbol, exchange, interval, start_date, end_date)` | OHLCV candles | DataFrame (timestamp, open, high, low, close, volume) |
| `client.quotes(symbol, exchange)` | Real-time snapshot | Dict (open, high, low, ltp, bid, ask, prev_close, volume) |
| `client.multiquotes(symbols=[...])` | Multi-symbol quotes | List of quote dicts |
| `client.depth(symbol, exchange)` | Market depth (L5) | Dict (bids, asks, ohlc, volume, oi) |
| `client.intervals()` | Available intervals | Dict (minutes, hours, days, weeks, months) |
| `client.optionchain(underlying, exchange, expiry_date, strike_count)` | Option chain around ATM | Dict (underlying_ltp, atm_strike, chain with ce/pe per strike) |
| `client.optiongreeks(symbol, exchange, interest_rate, ...)` | Option greeks + IV | Dict (greeks: delta/gamma/theta/vega/rho, implied_volatility, days_to_expiry) |
| `client.expiry(symbol, exchange, instrumenttype)` | Expiry dates list | Dict (data: list of expiry dates) |
| `client.connect()` | WebSocket connect | None (sets up WS connection) |
| `client.subscribe_ltp(instruments, callback)` | Live LTP stream | Callback with `{symbol, exchange, ltp}` |
| `client.subscribe_quote(instruments, callback)` | Live quote stream | Callback with `{symbol, exchange, ohlc, ltp, volume}` |
| `client.subscribe_depth(instruments, callback)` | Live depth stream | Callback with `{symbol, exchange, bids, asks}` |
## Indicator Library Reference
All indicators accessed via `from openalgo import ta`:
### Trend (20)
`ta.sma`, `ta.ema`, `ta.wma`, `ta.dema`, `ta.tema`, `ta.hma`, `ta.vwma`, `ta.alma`, `ta.kama`, `ta.zlema`, `ta.t3`, `ta.frama`, `ta.supertrend`, `ta.ichimoku`, `ta.chande_kroll_stop`, `ta.trima`, `ta.mcginley`, `ta.vidya`, `ta.alligator`, `ta.ma_envelopes`
### Momentum (9)
`ta.rsi`, `ta.macd`, `ta.stochastic`, `ta.cci`, `ta.williams_r`, `ta.bop`, `ta.elder_ray`, `ta.fisher`, `ta.crsi`
### Volatility (16)
`ta.atr`, `ta.bbands`, `ta.keltner`, `ta.donchian`, `ta.chaikin_volatility`, `ta.natr`, `ta.rvi`, `ta.ultimate_oscillator`, `ta.true_range`, `ta.massindex`, `ta.bb_percent`, `ta.bb_width`, `ta.chandelier_exit`, `ta.historical_volatility`, `ta.ulcer_index`, `ta.starc`
### Volume (15)
`ta.obv`, `ta.obv_smoothed`, `ta.vwap`, `ta.mfi`, `ta.adl`, `ta.cmf`, `ta.emv`, `ta.force_index`, `ta.nvi`, `ta.pvi`, `ta.volosc`, `ta.vroc`, `ta.kvo`, `ta.pvt`, `ta.rvol`
### Oscillators (20+)
`ta.cmo`, `ta.trix`, `ta.uo_oscillator`, `ta.awesome_oscillator`, `ta.accelerator_oscillator`, `ta.ppo`, `ta.po`, `ta.dpo`, `ta.aroon_oscillator`, `ta.stoch_rsi`, `ta.rvi_oscillator`, `ta.cho`, `ta.chop`, `ta.kst`, `ta.tsi`, `ta.vortex`, `ta.gator_oscillator`, `ta.stc`, `ta.coppock`, `ta.roc`
### Statistical (9)
`ta.linreg`, `ta.lrslope`, `ta.correlation`, `ta.beta`, `ta.variance`, `ta.tsf`, `ta.median`, `ta.mode`, `ta.median_bands`
### Hybrid (6+)
`ta.adx`, `ta.dmi`, `ta.aroon`, `ta.pivot_points`, `ta.sar`, `ta.williams_fractals`, `ta.rwi`
### TA-Lib Compatible (18, new in openalgo 2.0)
`ta.mom`, `ta.rocp`, `ta.rocr`, `ta.rocr100`, `ta.apo`, `ta.midpoint`, `ta.midprice`, `ta.avgprice`, `ta.medprice`, `ta.typprice`, `ta.wclprice`, `ta.plus_dm`, `ta.minus_dm`, `ta.dx`, `ta.adxr`, `ta.stochf`, `ta.linregangle`, `ta.linregintercept`
### Utilities
`ta.crossover`, `ta.crossunder`, `ta.cross`, `ta.highest`, `ta.lowest`, `ta.change`, `ta.roc`, `ta.stdev`, `ta.exrem`, `ta.flip`, `ta.valuewhen`, `ta.rising`, `ta.falling`
## Modular Rule Files
Detailed reference for each topic is in `rules/`:
| Rule File | Topic |
|-----------|-------|
| [indicator-catalog](rules/indicator-catalog.md) | Complete 100+ indicator reference with signatures and parameters |
| [data-fetching](rules/data-fetching.md) | OpenAlgo history/quotes/depth, yfinance, data normalization |
| [plotting](rules/plotting.md) | Plotly candlestick, overlay, subplot, multi-panel charts |
| [custom-indicators](rules/custom-indicators.md) | Building custom indicators with vectorized NumPy + ta primitives |
| [websocket-feeds](rules/websocket-feeds.md) | Real-time LTP/Quote/Depth streaming via WebSocket |
| [performance](rules/performance.md) | Rust core performance, O(n) guarantees, benchmarking |
| [dashboard-patterns](rules/dashboard-patterns.md) | Plotly Dash web applications with callbacks |
| [streamlit-patterns](rules/streamlit-patterns.md) | Streamlit web applications with sidebar, metrics, plotly charts |
| [multi-timeframe](rules/multi-timeframe.md) | Multi-timeframe indicator analysis |
| [signal-generation](rules/signal-generation.md) | Signal generation, cleaning, crossover/crossunder |
| [indicator-combinations](rules/indicator-combinations.md) | Combining indicators for confluence analysis |
| [symbol-format](rules/symbol-format.md) | OpenAlgo symbol format, exchange codes, index symbols |
## Chart Templates (in rules/assets/)
| Template | Path | Description |
|----------|------|-------------|
| EMA Chart | `assets/ema_chart/chart.py` | EMA overlay on candlestick |
| RSI Chart | `assets/rsi_chart/chart.py` | RSI with overbought/oversold zones |
| MACD Chart | `assets/macd_chart/chart.py` | MACD line, signal, histogram |
| Supertrend | `assets/supertrend_chart/chart.py` | Supertrend overlay with direction coloring |
| Bollinger | `assets/bollinger_chart/chart.py` | Bollinger Bands with squeeze detection |
| Multi-Indicator | `assets/multi_indicator/chart.py` | Candlestick + EMA + RSI + MACD + Volume |
| Basic Dashboard | `assets/dashboard_basic/app.py` | Single-symbol Plotly Dash app |
| Multi Dashboard | `assets/dashboard_multi/app.py` | Multi-symbol multi-timeframe dashboard |
| Streamlit Basic | `assets/streamlit_basic/app.py` | Single-symbol Streamlit app |
| Streamlit Multi | `assets/streamlit_multi/app.py` | Multi-timeframe Streamlit app |
| Custom Indicator | `assets/custom_indicator/template.py` | NumPy custom indicator template (composes ta primitives) |
| Live Feed | `assets/live_feed/template.py` | WebSocket real-time indicator |
| Scanner | `assets/scanner/template.py` | Multi-symbol indicator scanner |
## Quick Template: Standard Indicator Chart Script
```python
import os
from datetime import datetime, timedelta
from pathlib import Path
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta
# --- Config ---
script_dir = Path(__file__).resolve().parent
load_dotenv(find_dotenv(), override=False)
SYMBOL = "SBIN"
EXCHANGE = "NSE"
INTERVAL = "D"
# --- Fetch Data ---
client = api(
api_key=os.getenv("OPENALGO_API_KEY"),
host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
)
end_date = datetime.now().date()
start_date = end_date - timedelta(days=365)
df = client.history(
symbol=SYMBOL, exchange=EXCHANGE, interval=INTERVAL,
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)
if "timestamp" in df.columns:
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")
else:
df.index = pd.to_datetime(df.index)
df = df.sort_index()
if df.index.tz is not None:
df.index = df.index.tz_convert(None)
close = df["close"]
high = df["high"]
low = df["low"]
volume = df["volume"]
# --- Compute Indicators ---
ema_20 = ta.ema(close, 20)
rsi_14 = ta.rsi(close, 14)
# --- Chart ---
fig = make_subplots(
rows=2, cols=1, shared_xaxes=True,
row_heights=[0.7, 0.3], vertical_spacing=0.03,
subplot_titles=[f"{SYMBOL} Price + EMA(20)", "RSI(14)"],
)
# Candlestick
x_labels = df.index.strftime("%Y-%m-%d")
fig.add_trace(go.Candlestick(
x=x_labels, open=df["open"], high=high, low=low, close=close,
name="Price",
), row=1, col=1)
# EMA overlay
fig.add_trace(go.Scatter(
x=x_labels, y=ema_20, mode="lines",
name="EMA(20)", line=dict(color="cyan", width=1.5),
), row=1, col=1)
# RSI subplot
fig.add_trace(go.Scatter(
x=x_labels, y=rsi_14, mode="lines",
name="RSI(14)", line=dict(color="yellow", width=1.5),
), row=2, col=1)
fig.add_hline(y=70, line_dash="dash", line_color="red", row=2, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=2, col=1)
fig.update_layout(
template="plotly_dark", title=f"{SYMBOL} Technical Analysis",
xaxis_rangeslider_visible=False, xaxis_type="category",
xaxis2_type="category", height=700,
)
fig.show()
```