Alerts and ServLoci Dispatch

Generate de-duplicated alerts and pass reviewed intents to a dry-run ServLoci order boundary.

Part 23 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).")

Alert first, order later

This notebook turns completed candles into stateful alerts. It de-duplicates repeated signals, includes the observed values, and defaults to console output. Telegram and generic webhook sinks are opt-in. A separate, dry-run order boundary shows where ServLoci belongs.

!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

from dataclasses import dataclass
from datetime import datetime, timezone
import json, os, time, requests

@dataclass(frozen=True)
class Alert:
    key: str
    symbol: str
    side: str
    message: str
    observed_at: str

class AlertRouter:
    def __init__(self, cooldown_seconds=900):
        self.cooldown = cooldown_seconds
        self.sent_at = {}

    def should_send(self, alert):
        now = time.time(); previous = self.sent_at.get(alert.key, 0)
        if now - previous < self.cooldown:
            return False
        self.sent_at[alert.key] = now
        return True

    def console(self, alert):
        print(json.dumps(alert.__dict__, indent=2))

    def webhook(self, alert, url):
        response = requests.post(url, json={"text": alert.message, "alert": alert.__dict__}, timeout=10)
        response.raise_for_status()

    def telegram(self, alert, bot_token, chat_id):
        url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
        response = requests.post(url, json={"chat_id": chat_id, "text": alert.message}, timeout=10)
        response.raise_for_status()

def closed_candle_signal(symbol, bars, indicators):
    # Call only after the broker confirms the candle is closed.
    latest, previous = indicators.iloc[-1], indicators.iloc[-2]
    price = float(bars["close"].iloc[-1])
    crossed_up = previous["08_macd"] <= previous["09_macd_signal"] and latest["08_macd"] > latest["09_macd_signal"]
    risk_ok = 45 <= latest["14_rsi_14"] <= 70 and latest["37_adx_14"] >= 20
    if not (crossed_up and risk_ok):
        return None
    when = bars.index[-1].isoformat()
    return Alert(key=f"{symbol}:macd-up:{when}", symbol=symbol, side="BUY_WATCH",
                 message=f"{symbol}: MACD crossed up; close={price:.2f}, RSI={latest['14_rsi_14']:.1f}, ADX={latest['37_adx_14']:.1f}",
                 observed_at=when)

# Reproducible dry run. Replace `bars` with notebook 22's broker adapter output.
rng = np.random.default_rng(99); n = 340
close = pd.Series(20000 + np.r_[rng.normal(-3, 30, 300), rng.normal(25, 20, 40)].cumsum())
bars = pd.DataFrame({"open": close.shift().fillna(close.iloc[0]), "high": close+35,
                     "low": close-35, "close": close, "volume": rng.integers(100000, 900000, n)},
                    index=pd.date_range("2025-01-01", periods=n, freq="15min", tz="Asia/Kolkata"))
indicators = compute_top_50(bars)
alert = closed_candle_signal("NIFTY", bars, indicators)
router = AlertRouter()
if alert and router.should_send(alert):
    router.console(alert)
    if os.getenv("ALERT_WEBHOOK_URL"):
        router.webhook(alert, os.environ["ALERT_WEBHOOK_URL"])
    if os.getenv("TELEGRAM_BOT_TOKEN") and os.getenv("TELEGRAM_CHAT_ID"):
        router.telegram(alert, os.environ["TELEGRAM_BOT_TOKEN"], os.environ["TELEGRAM_CHAT_ID"])
else:
    print("No new completed-candle alert. This is a valid outcome.")

Optional ServLoci order boundary (dry-run by default)

DRY_RUN = True

def dispatch_after_risk_checks(alert, quantity, max_quantity=1):
    if alert is None: return None
    if quantity < 1 or quantity > max_quantity: raise ValueError("quantity rejected by local risk gate")
    intent = {"symbol": alert.symbol, "side": "BUY", "quantity": quantity,
              "client_order_id": alert.key, "signal_time": alert.observed_at}
    if DRY_RUN:
        print("DRY RUN — order intent not sent:", intent)
        return intent

    # Configure before constructing the supported broker SDK client:
    # from servloci import configure
    # configure(token=os.environ["STATIC_IP_TOKEN"], broker=os.environ["BROKER"])
    # broker = build_current_broker_client_from_private_secrets()
    # return broker.place_order(... current broker fields ...)
    raise RuntimeError("Wire one reviewed broker adapter before disabling DRY_RUN")

dispatch_after_risk_checks(alert, quantity=1)

Production alerts should persist de-duplication state outside the notebook, retry with bounded backoff, record delivery responses and expose a heartbeat/no-data alert. The order worker should independently re-check candle freshness, position, funds/margin, market session, maximum loss, quantity and the idempotency key. An alert is evidence to review—not proof of a profitable trade.

---

« Previous: Broker Data to Indicator Pipeline

Try the concepts above interactively: Options Strategy Builder · Docs · Get your static IP