""" AkDevz AI Market Intelligence — FYERS API V3 Broker Adapter Implements BaseBrokerAdapter using authentic FYERS REST client & rate limiter. """ import time import logging from typing import Dict, Any, List, Optional from python.brokers.base_broker import BaseBrokerAdapter from python.fyers.rate_limiter import FyersRateLimiter logger = logging.getLogger(__name__) class FyersBrokerAdapter(BaseBrokerAdapter): def __init__(self, credentials: Dict[str, Any]): super().__init__("FYERS", credentials) self.app_id = credentials.get("client_id") or credentials.get("app_id", "") self.secret_key = credentials.get("secret_key", "") self.access_token = credentials.get("access_token", "") self.rate_limiter = FyersRateLimiter() def authenticate(self) -> bool: if self.access_token and len(self.access_token) > 20: self.is_authenticated = True return True return False def get_quotes(self, symbols: List[str]) -> Dict[str, Dict[str, Any]]: if not self.access_token: return {} import requests self.rate_limiter.acquire() headers = { "Authorization": f"{self.app_id}:{self.access_token}", "Content-Type": "application/json" } url = f"https://api-t1.fyers.in/data/quotes?symbols={','.join(symbols[:50])}" try: resp = requests.get(url, headers=headers, timeout=8) if resp.status_code == 200: data = resp.json() quotes = {} for item in data.get("d", []): name = item.get("n") val = item.get("v", {}) if name and val: quotes[name] = { "symbol": name, "ltp": float(val.get("lp", 0.0)), "open": float(val.get("open_price", 0.0)), "high": float(val.get("high_price", 0.0)), "low": float(val.get("low_price", 0.0)), "close": float(val.get("prev_close_price", 0.0)), "volume": int(val.get("volume", 0)), "change_pct": float(val.get("chp", 0.0)), "timestamp": int(time.time()), "provider": "FYERS" } return quotes except Exception as e: logger.warning(f"FYERS quote fetch error: {e}") return {} def get_historical_candles( self, symbol: str, resolution: str = "5", range_from: str = "", range_to: str = "" ) -> List[List[Any]]: if not self.access_token: return [] import requests self.rate_limiter.acquire() headers = {"Authorization": f"{self.app_id}:{self.access_token}"} url = f"https://api-t1.fyers.in/data/history?symbol={symbol}&resolution={resolution}&date_format=1&range_from={range_from}&range_to={range_to}&cont_flag=1" try: resp = requests.get(url, headers=headers, timeout=12) if resp.status_code == 200: data = resp.json() return data.get("candles", []) except Exception as e: logger.warning(f"FYERS history error: {e}") return [] def get_option_chain(self, symbol: str, strike_count: int = 15) -> Optional[Dict[str, Any]]: if not self.access_token: return None import requests self.rate_limiter.acquire() headers = {"Authorization": f"{self.app_id}:{self.access_token}"} url = f"https://api-t1.fyers.in/data/options-chain-v3?symbol={symbol}&strikecount={strike_count}" try: resp = requests.get(url, headers=headers, timeout=12) if resp.status_code == 200: return resp.json().get("data") except Exception as e: logger.warning(f"FYERS option chain error: {e}") return None def get_funds_and_margin(self) -> Dict[str, Any]: if not self.access_token: return { "available_funds": 0.0, "used_margin": 0.0, "status": "UNAUTHENTICATED" } import requests self.rate_limiter.acquire() headers = {"Authorization": f"{self.app_id}:{self.access_token}"} url = "https://api-t1.fyers.in/api/v3/funds" try: resp = requests.get(url, headers=headers, timeout=8) if resp.status_code == 200: data = resp.json() fund_items = data.get("fund_limit", []) avail = 0.0 used = 0.0 for f in fund_items: if f.get("title") == "Total Balance": avail = float(f.get("equityAmount", 0.0)) elif f.get("title") == "Utilized Amount": used = float(f.get("equityAmount", 0.0)) return { "available_funds": avail, "used_margin": used, "status": "CONNECTED", "provider": "FYERS" } except Exception as e: logger.warning(f"FYERS funds fetch error: {e}") return {"available_funds": 0.0, "used_margin": 0.0, "status": "ERROR"} def get_positions(self) -> List[Dict[str, Any]]: if not self.access_token: return [] import requests self.rate_limiter.acquire() headers = {"Authorization": f"{self.app_id}:{self.access_token}"} url = "https://api-t1.fyers.in/api/v3/positions" try: resp = requests.get(url, headers=headers, timeout=8) if resp.status_code == 200: data = resp.json() return data.get("netPositions", []) except Exception as e: logger.warning(f"FYERS positions fetch error: {e}") return [] def place_order(self, order_params: Dict[str, Any]) -> Dict[str, Any]: """Submits verified order to FYERS API V3.""" if not self.access_token: return {"status": "error", "message": "FYERS access token not present"} import requests self.rate_limiter.acquire() headers = { "Authorization": f"{self.app_id}:{self.access_token}", "Content-Type": "application/json" } url = "https://api-t1.fyers.in/api/v3/orders/sync" try: resp = requests.post(url, headers=headers, json=order_params, timeout=10) return resp.json() except Exception as e: return {"status": "error", "message": str(e)} def modify_order(self, order_id: str, params: Dict[str, Any]) -> Dict[str, Any]: if not self.access_token: return {"status": "error", "message": "Unauthenticated"} import requests self.rate_limiter.acquire() headers = {"Authorization": f"{self.app_id}:{self.access_token}", "Content-Type": "application/json"} url = "https://api-t1.fyers.in/api/v3/orders/sync" payload = {"id": order_id, **params} try: resp = requests.put(url, headers=headers, json=payload, timeout=10) return resp.json() except Exception as e: return {"status": "error", "message": str(e)} def cancel_order(self, order_id: str) -> Dict[str, Any]: if not self.access_token: return {"status": "error", "message": "Unauthenticated"} import requests self.rate_limiter.acquire() headers = {"Authorization": f"{self.app_id}:{self.access_token}", "Content-Type": "application/json"} url = "https://api-t1.fyers.in/api/v3/orders/sync" try: resp = requests.delete(url, headers=headers, json={"id": order_id}, timeout=10) return resp.json() except Exception as e: return {"status": "error", "message": str(e)}