ORB Trading System

# ===================================================== # FULL STACK ORB + SHOONYA SYSTEM (1 FILE ONLY) # BACKEND + FRONTEND COMBINED # ===================================================== from fastapi import FastAPI from fastapi.responses import HTMLResponse import pyotp from NorenRestApiPy.NorenApi import NorenApi app = FastAPI() # ===================================================== # SHOONYA API # ===================================================== class ShoonyaApi(NorenApi): def __init__(self): super().__init__( host="https://api.shoonya.com/NorenWClientTP/", websocket="wss://api.shoonya.com/NorenWSTP/" ) api = ShoonyaApi() # ===================================================== # GLOBAL STATE # ===================================================== state = { "logged_in": False, "symbols": { "NIFTY": {"active": False, "orb": None}, "BANKNIFTY": {"active": False, "orb": None}, "SENSEX": {"active": False, "orb": None} }, "trades": [], "payoff": {} } # ===================================================== # ORB ENGINE # ===================================================== class ORB: def __init__(self): self.high = None self.low = None self.done = False def update(self, price): self.high = price if self.high is None else max(self.high, price) self.low = price if self.low is None else min(self.low, price) def signal(self, price): if self.done or self.high is None: return None mid = (self.high + self.low) / 2 rng = self.high - self.low upper = mid + (rng * 0.14) lower = mid - (rng * 0.14) if price > upper: self.done = True return "BUY" if price < lower: self.done = True return "SELL" return None # ===================================================== # LOGIN (REAL SHOONYA) # ===================================================== @app.post("/login") def login(data: dict): totp = pyotp.TOTP(data["totp"]).now() res = api.login( userid=data["user"], password=data["password"], twoFA=totp, vendor_code=data["vendor"], api_secret=data["api_key"], imei="123456" ) if res and res.get("stat") == "Ok": state["logged_in"] = True return {"status": "OK"} return {"status": "FAIL"} # ===================================================== # ORDER EXECUTION # ===================================================== def place_order(symbol, side): return api.place_order( buy_or_sell="B" if side == "BUY" else "S", product_type="MIS", exchange="NSE", tradingsymbol=symbol, quantity=1, price_type="MKT" ) # ===================================================== # PROCESS TICK (CORE ENGINE) # ===================================================== def process(symbol, price): if not state["symbols"][symbol]["active"]: return if state["symbols"][symbol]["orb"] is None: state["symbols"][symbol]["orb"] = ORB() orb = state["symbols"][symbol]["orb"] orb.update(price) signal = orb.signal(price) if signal: order = place_order(symbol, signal) state["trades"].append({ "symbol": symbol, "signal": signal, "price": price, "order": str(order), "status": "EXECUTED" }) # ===================================================== # START / STOP # ===================================================== @app.post("/start") def start(data: dict): s = data["symbol"] state["symbols"][s]["active"] = True state["symbols"][s]["orb"] = ORB() return {"status": "STARTED"} @app.post("/stop") def stop(data: dict): state["symbols"][data["symbol"]]["active"] = False return {"status": "STOPPED"} # ===================================================== # EXIT ALL # ===================================================== @app.post("/exit_all") def exit_all(): for s in state["symbols"]: api.place_order( buy_or_sell="S", product_type="MIS", exchange="NSE", tradingsymbol=s, quantity=1, price_type="MKT" ) return {"status": "EXITED"} # ===================================================== # FRONTEND UI (SAME FILE) # ===================================================== @app.get("/") def home(): return HTMLResponse(""" ORB Trading System

LIVE ORB TRADING DASHBOARD

LOGIN

SYMBOL CONTROL

TRADE TABLE

""") # ===================================================== # TRADE VIEW API # ===================================================== @app.get("/trades") def trades(): return state["trades"]

Pulse ORB Terminal