Broker Data to Indicator Pipeline¶
Normalize broker OHLCV, compute all 50 indicators, and keep reads separate from routed order calls.
Part 22 of 24 in the ServLoci algo/options trading notebook series — full index in notebooks/README.md.
Setup¶
# Get your dedicated static IPv6 + SOCKS5 credentials free:
# https://comm.servloci.in/register (or /auth/google?free=1 for an instant trial)
# Your api_key / api_secret pair shows up in the portal after signup:
# https://comm.servloci.in/user
!pip install -q "requests[socks]"
!curl -sL https://comm.servloci.in/sdk/servloci.py -o servloci.py
import os
from servloci import ServLoci
SERVLOCI_API_KEY = os.environ.get("SERVLOCI_API_KEY", "dhan:1000000001") # broker:client_id
SERVLOCI_API_SECRET = os.environ.get("SERVLOCI_API_SECRET", "") # from the portal — leave blank to run this notebook in demo mode
sl = None
if SERVLOCI_API_SECRET:
sl = ServLoci(api_key=SERVLOCI_API_KEY, api_secret=SERVLOCI_API_SECRET)
print("ServLoci configured:", sl.host, sl.port)
else:
print("SERVLOCI_API_SECRET not set — running in demo mode (no live proxy calls).")
One normalized shape for every broker¶
Broker payloads differ, but indicators need only timestamp, open, high, low, close, volume. Keep a small fetch adapter per broker, normalize immediately, and make the rest of the research code broker-neutral. Market-data reads stay direct; create the ServLoci-routed SDK client only at the order boundary.
!pip install -q pandas numpy requests
import numpy as np
import pandas as pd
def compute_top_50(frame):
"""Return 50 named indicators from an OHLCV DataFrame.
Input columns are case-insensitive: open, high, low, close and volume.
Warm-up rows contain NaN by design; never backfill them into a live signal.
"""
df = frame.rename(columns={c: str(c).lower() for c in frame.columns}).copy()
required = {"open", "high", "low", "close", "volume"}
missing = required.difference(df.columns)
if missing:
raise ValueError(f"missing OHLCV columns: {sorted(missing)}")
o, h, l, c, v = (df[x].astype(float) for x in ("open", "high", "low", "close", "volume"))
out = pd.DataFrame(index=df.index)
safe = lambda x: x.replace([np.inf, -np.inf], np.nan)
ema = lambda x, n: x.ewm(span=n, adjust=False, min_periods=n).mean()
wma = lambda x, n: x.rolling(n).apply(
lambda a: np.dot(a, np.arange(1, n + 1)) / (n * (n + 1) / 2), raw=True
)
# Trend and moving-average family (1-11)
out["01_sma_20"] = c.rolling(20).mean()
out["02_ema_20"] = ema(c, 20)
out["03_wma_20"] = wma(c, 20)
out["04_hma_20"] = wma(2 * wma(c, 10) - wma(c, 20), 4)
e1 = ema(c, 20); e2 = ema(e1, 20); e3 = ema(e2, 20)
out["05_dema_20"] = 2 * e1 - e2
out["06_tema_20"] = 3 * e1 - 3 * e2 + e3
out["07_vwma_20"] = safe((c * v).rolling(20).sum() / v.rolling(20).sum())
macd = ema(c, 12) - ema(c, 26)
out["08_macd"] = macd
out["09_macd_signal"] = ema(macd, 9)
out["10_ppo"] = safe(100 * macd / ema(c, 26))
ex1 = ema(c, 15); ex2 = ema(ex1, 15); ex3 = ema(ex2, 15)
out["11_trix"] = ex3.pct_change(fill_method=None) * 100
# Momentum and oscillator family (12-24)
out["12_roc_12"] = c.pct_change(12, fill_method=None) * 100
out["13_momentum_10"] = c.diff(10)
delta = c.diff(); gain = delta.clip(lower=0); loss = -delta.clip(upper=0)
avg_gain = gain.ewm(alpha=1/14, adjust=False, min_periods=14).mean()
avg_loss = loss.ewm(alpha=1/14, adjust=False, min_periods=14).mean()
out["14_rsi_14"] = 100 - (100 / (1 + safe(avg_gain / avg_loss)))
low14, high14 = l.rolling(14).min(), h.rolling(14).max()
stoch = safe(100 * (c - low14) / (high14 - low14))
out["15_stochastic_k"] = stoch
out["16_stochastic_d"] = stoch.rolling(3).mean()
out["17_williams_r"] = safe(-100 * (high14 - c) / (high14 - low14))
typical = (h + l + c) / 3
tp_mean = typical.rolling(20).mean()
mean_dev = typical.rolling(20).apply(lambda a: np.mean(np.abs(a - a.mean())), raw=True)
out["18_cci_20"] = safe((typical - tp_mean) / (0.015 * mean_dev))
prev = c.shift(1)
buy_pressure = c - pd.concat([l, prev], axis=1).min(axis=1)
true_range = pd.concat([h - l, (h - prev).abs(), (l - prev).abs()], axis=1).max(axis=1)
out["19_ultimate_oscillator"] = safe(100 * (
4 * buy_pressure.rolling(7).sum() / true_range.rolling(7).sum()
+ 2 * buy_pressure.rolling(14).sum() / true_range.rolling(14).sum()
+ buy_pressure.rolling(28).sum() / true_range.rolling(28).sum()
) / 7)
midpoint = (h + l) / 2
out["20_awesome_oscillator"] = midpoint.rolling(5).mean() - midpoint.rolling(34).mean()
r1, r2, r3, r4 = (c.pct_change(n, fill_method=None) * 100 for n in (10, 15, 20, 30))
out["21_kst"] = r1.rolling(10).sum() + 2*r2.rolling(10).sum() + 3*r3.rolling(10).sum() + 4*r4.rolling(15).sum()
pc = c.diff(); apc = pc.abs()
out["22_tsi"] = safe(100 * ema(ema(pc, 25), 13) / ema(ema(apc, 25), 13))
streak = pd.Series(0.0, index=c.index)
for i in range(1, len(c)):
direction = np.sign(c.iloc[i] - c.iloc[i - 1])
prior = streak.iloc[i - 1]
streak.iloc[i] = 0 if direction == 0 else direction * (abs(prior) + 1 if np.sign(prior) == direction else 1)
def rsi_series(x, n):
d = x.diff(); g = d.clip(lower=0).ewm(alpha=1/n, adjust=False, min_periods=n).mean()
q = (-d.clip(upper=0)).ewm(alpha=1/n, adjust=False, min_periods=n).mean()
return 100 - 100 / (1 + safe(g / q))
pct_rank = c.pct_change(fill_method=None).rolling(100).apply(lambda a: 100 * (a[-1] > a[:-1]).mean(), raw=True)
out["23_connors_rsi"] = (rsi_series(c, 3) + rsi_series(streak, 2) + pct_rank) / 3
sum_gain, sum_loss = gain.rolling(14).sum(), loss.rolling(14).sum()
out["24_cmo_14"] = safe(100 * (sum_gain - sum_loss) / (sum_gain + sum_loss))
# Volatility and channel family (25-36)
mid = c.rolling(20).mean(); std = c.rolling(20).std(ddof=0)
upper, lower = mid + 2*std, mid - 2*std
out["25_bollinger_upper"] = upper
out["26_bollinger_lower"] = lower
out["27_bollinger_bandwidth"] = safe(100 * (upper - lower) / mid)
atr = true_range.ewm(alpha=1/14, adjust=False, min_periods=14).mean()
out["28_atr_14"] = atr
out["29_natr_14"] = safe(100 * atr / c)
out["30_true_range"] = true_range
kel_mid = ema(c, 20)
out["31_keltner_upper"] = kel_mid + 2 * atr
out["32_keltner_lower"] = kel_mid - 2 * atr
out["33_donchian_upper"] = h.rolling(20).max()
out["34_donchian_lower"] = l.rolling(20).min()
out["35_stddev_20"] = std
out["36_historical_volatility"] = np.log(c / c.shift(1)).rolling(20).std(ddof=0) * np.sqrt(252) * 100
# Directional, stop and cloud family (37-46)
up_move, down_move = h.diff(), -l.diff()
plus_dm = up_move.where((up_move > down_move) & (up_move > 0), 0.0)
minus_dm = down_move.where((down_move > up_move) & (down_move > 0), 0.0)
plus_di = safe(100 * plus_dm.ewm(alpha=1/14, adjust=False, min_periods=14).mean() / atr)
minus_di = safe(100 * minus_dm.ewm(alpha=1/14, adjust=False, min_periods=14).mean() / atr)
out["37_adx_14"] = safe(100 * (plus_di - minus_di).abs() / (plus_di + minus_di)).ewm(alpha=1/14, adjust=False, min_periods=14).mean()
out["38_plus_di"] = plus_di
out["39_minus_di"] = minus_di
out["40_aroon_up"] = h.rolling(25).apply(lambda a: 100 * (np.argmax(a) + 1) / len(a), raw=True)
out["41_aroon_down"] = l.rolling(25).apply(lambda a: 100 * (np.argmin(a) + 1) / len(a), raw=True)
out["42_vortex_plus"] = safe((h - l.shift(1)).abs().rolling(14).sum() / true_range.rolling(14).sum())
out["43_vortex_minus"] = safe((l - h.shift(1)).abs().rolling(14).sum() / true_range.rolling(14).sum())
psar = pd.Series(np.nan, index=c.index)
if len(c):
bull, af, extreme = True, 0.02, h.iloc[0]
psar.iloc[0] = l.iloc[0]
for i in range(1, len(c)):
candidate = psar.iloc[i-1] + af * (extreme - psar.iloc[i-1])
if bull:
candidate = min(candidate, l.iloc[i-1], l.iloc[max(i-2, 0)])
if l.iloc[i] < candidate: bull, candidate, af, extreme = False, extreme, 0.02, l.iloc[i]
elif h.iloc[i] > extreme: extreme, af = h.iloc[i], min(af + 0.02, 0.2)
else:
candidate = max(candidate, h.iloc[i-1], h.iloc[max(i-2, 0)])
if h.iloc[i] > candidate: bull, candidate, af, extreme = True, extreme, 0.02, h.iloc[i]
elif l.iloc[i] < extreme: extreme, af = l.iloc[i], min(af + 0.02, 0.2)
psar.iloc[i] = candidate
out["44_parabolic_sar"] = psar
out["45_ichimoku_conversion"] = (h.rolling(9).max() + l.rolling(9).min()) / 2
out["46_ichimoku_base"] = (h.rolling(26).max() + l.rolling(26).min()) / 2
# Volume and money-flow family (47-50)
out["47_obv"] = (np.sign(c.diff()).fillna(0) * v).cumsum()
raw_money = typical * v; positive = raw_money.where(typical.diff() > 0, 0); negative = raw_money.where(typical.diff() < 0, 0)
out["48_mfi_14"] = 100 - 100 / (1 + safe(positive.rolling(14).sum() / negative.rolling(14).sum()))
money_flow_multiplier = safe(((c - l) - (h - c)) / (h - l))
money_flow_volume = money_flow_multiplier * v
out["49_cmf_20"] = safe(money_flow_volume.rolling(20).sum() / v.rolling(20).sum())
out["50_accumulation_distribution"] = money_flow_volume.fillna(0).cumsum()
assert out.shape[1] == 50
return out
def normalize_ohlcv(records, mapping):
raw = pd.DataFrame(records).rename(columns={source: target for target, source in mapping.items()})
needed = ["timestamp", "open", "high", "low", "close", "volume"]
missing = set(needed).difference(raw.columns)
if missing:
raise ValueError(f"adapter did not provide: {sorted(missing)}")
raw["timestamp"] = pd.to_datetime(raw["timestamp"], utc=True)
bars = raw.set_index("timestamp")[["open", "high", "low", "close", "volume"]].sort_index()
return bars.apply(pd.to_numeric, errors="raise")
# Generic example: replace this callback with a broker SDK/API fetch.
def demo_fetch():
rng = np.random.default_rng(12); n = 320
close = pd.Series(24000 + rng.normal(0, 60, n).cumsum())
return pd.DataFrame({
"time": pd.date_range("2025-01-01", periods=n, freq="B", tz="Asia/Kolkata"),
"o": close.shift(1).fillna(close.iloc[0]), "h": close + 50,
"l": close - 50, "c": close, "v": rng.integers(100000, 800000, n),
}).to_dict("records")
bars = normalize_ohlcv(demo_fetch(), {
"timestamp": "time", "open": "o", "high": "h", "low": "l", "close": "c", "volume": "v"
})
features = compute_top_50(bars)
dataset = bars.join(features)
display(dataset.tail(3))
Concrete broker fetch adapters¶
# Zerodha Kite (direct read; create `kite` with its normal authenticated flow)
def fetch_kite_daily(kite, instrument_token, start, end):
rows = kite.historical_data(instrument_token, start, end, "day", continuous=False, oi=False)
return normalize_ohlcv(rows, {
"timestamp": "date", "open": "open", "high": "high", "low": "low", "close": "close", "volume": "volume"
})
# FYERS v3 (direct read; `fyers.history` returns candles as [epoch,o,h,l,c,v])
def fetch_fyers_daily(fyers, symbol, start_epoch, end_epoch):
response = fyers.history(data={"symbol": symbol, "resolution": "D", "date_format": "0",
"range_from": str(start_epoch), "range_to": str(end_epoch), "cont_flag": "1"})
rows = [dict(zip(["timestamp", "open", "high", "low", "close", "volume"], row))
for row in response.get("candles", [])]
return normalize_ohlcv(rows, {name: name for name in rows[0]} if rows else {
"timestamp": "timestamp", "open": "open", "high": "high", "low": "low", "close": "close", "volume": "volume"
})
# DhanHQ v2 REST (direct read). Security ID and segment come from Dhan's instrument master.
def fetch_dhan_daily(client_id, access_token, security_id, from_date, to_date):
import requests
response = requests.post("https://api.dhan.co/v2/charts/historical", headers={
"access-token": access_token, "client-id": client_id, "Content-Type": "application/json"
}, json={"securityId": str(security_id), "exchangeSegment": "IDX_I", "instrument": "INDEX",
"expiryCode": 0, "oi": False, "fromDate": from_date, "toDate": to_date}, timeout=30)
response.raise_for_status(); payload = response.json()
rows = [dict(zip(["timestamp", "open", "high", "low", "close", "volume"], values))
for values in zip(payload["timestamp"], payload["open"], payload["high"], payload["low"], payload["close"], payload["volume"])]
return normalize_ohlcv(rows, {k: k for k in ("timestamp", "open", "high", "low", "close", "volume")})
For Upstox, Groww, Breeze, Kotak Neo and every broker in notebook 20, implement only the fetch callback using that broker’s current SDK, then pass its records through normalize_ohlcv() and compute_top_50(). Keeping this boundary small makes schema changes visible and testable.
Cache raw candles with broker, symbol, interval, timezone, fetch time and adjustment policy. Do not silently mix feeds: two vendors can close the same candle differently because of session and timestamp rules.
« Previous: Top 50 Technical Indicators
Next: Alerts and ServLoci Dispatch »
Try the concepts above interactively: Options Strategy Builder · Docs · Get your static IP