Skip to content

Instantly share code, notes, and snippets.

@MooneDrJune

MooneDrJune/kiteconnect_extras.py

Created September 14, 2024 09:20
Show Gist options
  • Save MooneDrJune/4e4a8ef8b45eaf2cf356cf94c3dd4783 to your computer and use it in GitHub Desktop.
Save MooneDrJune/4e4a8ef8b45eaf2cf356cf94c3dd4783 to your computer and use it in GitHub Desktop.
Download ZIP
KiteConnect Extras Async KiteConnect
Raw
kiteconnect_extras.py
from __future__ import annotations
# -*- coding: utf-8 -*-
"""
:description: KiteConnect (Batteries Included) On Steroids Version.
:license: MIT.
:author: Dr June Moone
:created: On Saturday July 29, 2023 19:56:53 GMT+05:30
"""
__author__ = "Dr June Moone"
__webpage__ = "https://github.com/MooneDrJune"
__license__ = "MIT"
import os, gc, sys, copy, json, time, signal, pprint, pickle, socket, struct, asyncio, logging, urllib.parse, platform, contextlib, pandas as pd, dateutil.parser
from pathlib import Path
from functools import cache
from collections import deque
from itertools import takewhile
from json import JSONDecodeError
from types import SimpleNamespace
from pickle import HIGHEST_PROTOCOL
from functools import wraps, partial
from time import perf_counter, sleep
from datetime import datetime, timedelta
from aiohttp.resolver import AsyncResolver
from threading import Event, Thread, Timer
from multiprocessing import cpu_count, resource_tracker
from multiprocessing.shared_memory import SharedMemory, ShareableList
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
from typing import (
Any,
Callable,
Dict,
Tuple,
List,
Optional,
Union,
Literal,
NoReturn,
Set,
)
try:
import six, pyotp, aiohttp, requests, kiteconnect
import kiteconnect.exceptions as ex
from dill import dump, load
from six.moves.urllib.parse import urljoin
from requests.models import PreparedRequest
from kiteconnect import KiteConnect, KiteTicker
except (ImportError, ModuleNotFoundError):
os.system(
f"{sys.executable} -m pip install --upgrade dill"
+ " six pyotp requests aiohttp aiodns cchardet"
+ " kiteconnect websocket-client websockets"
)
import six, pyotp, aiohttp, requests, kiteconnect
import kiteconnect.exceptions as ex
from dill import dump, load
from six.moves.urllib.parse import urljoin
from requests.models import PreparedRequest
from kiteconnect import KiteConnect, KiteTicker
try:
# unix / macos only
from signal import SIGABRT, SIGINT, SIGTERM, SIGHUP
SIGNALS = (SIGABRT, SIGINT, SIGTERM, SIGHUP)
except (ImportError, ModuleNotFoundError):
from signal import SIGABRT, SIGINT, SIGTERM
SIGNALS = (SIGABRT, SIGINT, SIGTERM)
if not platform.system().lower().startswith("win") and sys.version_info >= (
3,
8,
): # noqa E501
try:
import uvloop
except (ImportError, ModuleNotFoundError):
os.system(f"{sys.executable} -m pip install uvloop")
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
else:
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
log = logging.getLogger(__name__)
class KiteExtUtils:
@staticmethod
def nbytes(frame: Any) -> int:
if isinstance(frame, (bytes, bytearray)):
return len(frame)
else:
try:
return frame.nbytes
except AttributeError:
return len(frame)
@staticmethod
def pack_frames_prelude(frames: Any) -> bytes:
lengths = [struct.pack("Q", len(frames))] + [
struct.pack("Q", KiteExtUtils.nbytes(frame)) for frame in frames
]
return b"".join(lengths)
@staticmethod
def pack_frames(frames: Any) -> Tuple[int, List[bytes]]:
prelude = [KiteExtUtils.pack_frames_prelude(frames)]
if not isinstance(frames, list):
frames = list(frames)
data_ls = prelude + frames
data_sz = sum(map(lambda b: len(b), data_ls))
return data_sz, data_ls
@staticmethod
def unpack_frames(bytes_array: Any) -> List[Any]:
frames = []
(n_frames,) = struct.unpack("Q", bytes_array[:8])
start = 8 + n_frames * 8
for i in range(n_frames):
(length,) = struct.unpack(
"Q", bytes_array[(i + 1) * 8 : (i + 2) * 8]
) # noqa E501
frame = bytes_array[start : start + length]
frames.append(frame)
start += length
return frames
@staticmethod
def remove_shm_from_resource_tracker() -> None:
def fix_register(name, rtype):
if rtype == "shared_memory":
return
return resource_tracker._resource_tracker.register(name, rtype) # noqa E501
resource_tracker.register = fix_register
def fix_unregister(name, rtype):
if rtype == "shared_memory":
return
return resource_tracker._resource_tracker.unregister(
name, rtype
) # noqa E501
resource_tracker.unregister = fix_unregister
if "shared_memory" in resource_tracker._CLEANUP_FUNCS: # type: ignore
del resource_tracker._CLEANUP_FUNCS["shared_memory"] # type: ignore
@staticmethod
def create_shm_list(
list_object: list, name: str
) -> Optional[ShareableList]: # noqa E501
KiteExtUtils.remove_shm_from_resource_tracker()
try:
return ShareableList(list_object, name=name)
except FileExistsError:
shared_list_object = KiteExtUtils.get_shm_list(name)
if shared_list_object is not None:
KiteExtUtils.close_shm_list(shared_list_object, unlink=True)
del shared_list_object
return ShareableList(list_object, name=name)
@staticmethod
def get_shm_list(name: str) -> Optional[ShareableList]:
KiteExtUtils.remove_shm_from_resource_tracker()
try:
return ShareableList(name=name)
except FileNotFoundError as e:
raise FileNotFoundError(
f"The Shared Memory At The Provided Namesapce: {name}"
+ "Does Not Exists, Please Check Whether The Shared Memory Exists, And If It Does, Please Pass The Correct Namesapce And Try Again." # noqa: E501
) from e
@staticmethod
def close_shm_list(
shared_list_object: ShareableList, unlink: bool = False
) -> None: # noqa E501
shared_list_object.shm.close()
if unlink:
shared_list_object.shm.unlink()
@staticmethod
def create_shm(
name: str = "testshm", size: int = 10485760
) -> SharedMemory: # noqa E501
KiteExtUtils.remove_shm_from_resource_tracker()
try:
return SharedMemory(name=name, create=True, size=size)
except FileExistsError:
shm = KiteExtUtils.read_shm(name)
if shm is not None:
KiteExtUtils.close_shm(shm, unlink=True)
del shm
return SharedMemory(name=name, create=True, size=size)
@staticmethod
def read_shm(name: str = "testshm") -> SharedMemory:
KiteExtUtils.remove_shm_from_resource_tracker()
try:
return SharedMemory(name=name)
except FileNotFoundError as e:
raise FileNotFoundError(
f"The Shared Memory At The Provided Namesapce: {name}"
+ "Does Not Exists, Please Check Whether The Shared Memory Exists, And If It Does, Please Pass The Correct Namesapce And Try Again." # noqa: E501
) from e
@staticmethod
def close_shm(shm: SharedMemory, unlink: bool = False) -> None:
shm.close()
if unlink:
shm.unlink()
@staticmethod
def shm_send_obj(obj: Any, shm: SharedMemory) -> int:
buffers = []
# Pack the buffers to be written to memory
data_sz, data_ls = KiteExtUtils.pack_frames(
[
pickle.dumps(
obj,
protocol=pickle.HIGHEST_PROTOCOL,
buffer_callback=lambda b: buffers.append(b.raw()),
)
]
+ buffers
)
# Write to shared memory
write_offset = 0
for data in data_ls:
write_end = write_offset + len(data)
shm.buf[write_offset:write_end] = data # type: ignore
write_offset = write_end
return data_sz
@staticmethod
def shm_recv_obj(shm: SharedMemory, size: int = 10485760) -> Any:
data = shm.buf[:size]
# Unpack and un-pickle the data buffers
buffers = KiteExtUtils.unpack_frames(data)
obj = pickle.loads(buffers[0], buffers=buffers[1:])
# Bring the `obj` out of shared memory
ret = copy.deepcopy(obj)
# Clean up
del data
del buffers
del obj
return ret
@staticmethod
def validate_object_path(object_path: Union[Path, str]) -> Path:
return (
Path(
f"{object_path.absolute()}"
if ".pkl" in str(object_path.absolute())
else f"{object_path.absolute()}.pkl"
)
if isinstance(object_path, Path)
else Path(
Path(
f"{object_path}"
if object_path.find(".pkl") != -1
else f"{object_path}.pkl"
).absolute() # noqa: E501
) # noqa: E501
)
@staticmethod
def save_object(
objects: Any,
object_path: Union[Path, str],
) -> None:
object_path = KiteExtUtils.validate_object_path(object_path)
if (
isinstance(object_path, Path)
and object_path.exists()
and object_path.is_file()
):
try:
with open(object_path, "wb") as output_file:
dump(objects, output_file, protocol=HIGHEST_PROTOCOL)
print(
f"{object_path} State Saved in {object_path} File"
) # noqa E501
except Exception as err:
print(
f"Failed To Save {object_path} State "
+ f"In {object_path}.pkl File"
+ f"\nAnd The Exception Was: {err}"
)
raise err
else:
raise ValueError(
"The Object Path Should be either of type pathlib.Path or str"
) # noqa: E501
@staticmethod
def read_object(object_path: Union[Path, str]) -> Any:
object_path = KiteExtUtils.validate_object_path(object_path)
if (
isinstance(object_path, Path)
and object_path.exists()
and object_path.is_file()
):
try:
with open(object_path, "rb") as input_file:
objects = load(input_file)
print(
f"{object_path} State Read From {object_path} File"
) # noqa E501
except Exception as err:
print(
f"Failed To Read {object_path} State"
+ f"From {object_path}.pkl File"
+ f"\nAnd The Exception Was: {err}"
)
raise err
else:
return objects
else:
raise ValueError(
"The Object Path Should be either of type pathlib.Path or str"
) # noqa: E501
class ProgramKilled(Exception):
"""ProgramKilled Checks the ProgramKilled exception"""
pass # type: ignore
class CustomTimer(Thread):
def __init__(
self,
name: str,
interval: float,
function: Callable,
args=None,
kwargs=None, # noqa E501
) -> None: # noqa E501
Thread.__init__(self=self)
self.interval: float = interval
self.function = function
self.args = args if args is not None else []
self.kwargs = kwargs if kwargs is not None else {}
self.finished = Event()
self.start_time: float = perf_counter()
self.next_iteration = perf_counter() + self.interval
self.daemon = True
self.name = name
self.first_run = True
def signal_handler(self, signum, frame) -> NoReturn:
self.cancel()
raise ProgramKilled
def cancel(self) -> None:
self.finished.set()
self.join(0.5)
def run(self) -> None:
with contextlib.suppress(ValueError):
for sig in SIGNALS:
signal.signal(signalnum=sig, handler=self.signal_handler)
while not self.finished.is_set():
try:
if perf_counter() >= self.next_iteration or self.first_run:
self.next_iteration: float = perf_counter() + self.interval
try:
self.function(*self.args, **self.kwargs)
except (ProgramKilled, KeyboardInterrupt, SystemExit):
self.finished.set()
break
if self.first_run:
self.first_run = False
if self.finished.is_set():
break
except (ProgramKilled, KeyboardInterrupt, SystemExit):
self.cancel()
break
sleep(0.005)
else:
self.finished.set()
class KiteExtTicker(KiteTicker):
def __graceful_exit(self) -> None:
if self.share_ticks:
self.publish_ticks.cancel()
self.fetch_rpcs.cancel()
self.process_rpcs.cancel()
if self.with_shared_ticks:
self.subscribe_ticks.cancel()
if hasattr(self, "ticks_shm") and self.ticks_shm is not None:
KiteExtUtils.close_shm(self.ticks_shm, unlink=self.share_ticks)
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.close_shm(self.rpc_shm, unlink=self.share_ticks)
def __enter__(self) -> "KiteExtTicker":
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.__graceful_exit()
def __del__(self) -> None:
self.__graceful_exit()
def __delete__(self) -> None:
self.__graceful_exit()
def handle_stop_signals(self, *args, **kwargs):
try:
self.__graceful_exit()
except Exception as err:
log.error(str(err))
else:
raise SystemExit
def receive_rpc(self) -> None:
try:
rpc_data = KiteExtUtils.shm_recv_obj(self.rpc_shm)
except IndexError:
pass
else:
self.rpc_calls_q.append(rpc_data)
def process_rpc(self) -> None:
if len(self.rpc_calls_q) == 0:
return
while len(self.rpc_calls_q) != 0:
fn, args, kwargs = self.rpc_calls_q.popleft()
try:
if hasattr(self, fn):
fn_call_result = getattr(self, fn)(*args, **kwargs)
except Exception as e:
log.error(
f"Exception Encountered While executing remote procedure call. Exception Was: {e}"
)
def initialize_fetch_rpc_calls(self) -> None:
self.rpc_calls_q = deque()
self.rpc_shm = KiteExtUtils.create_shm(name="KiteExtTickerRpc")
self.fetch_rpcs = CustomTimer(
"receive_rpc_calls",
0.05,
self.receive_rpc,
)
self.process_rpcs = CustomTimer(
"process_rpc_calls",
0.05,
self.process_rpc,
)
self.fetch_rpcs.start()
self.process_rpcs.start()
def publish_ticks_over_shm(self) -> None:
self.ticks_shm = KiteExtUtils.create_shm(name="KiteExtTicker")
self.publish_ticks = CustomTimer(
"publish_tickers_timer",
0.15,
KiteExtUtils.shm_send_obj,
(self.ticks, self.ticks_shm),
)
self.publish_ticks.start()
def subscribe(self, *args, **kwargs):
"""
Subscribe to a list of instrument_tokens.
- `instrument_tokens` is list of instrument instrument_tokens to subscribe
"""
if self.with_shared_ticks:
fn = sys._getframe().f_code.co_name
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.shm_send_obj((fn, args, kwargs), self.rpc_shm)
self.subscribed_tokens.update(set(args[0]))
else:
try:
instrument_tokens = args[0]
self.ws.sendMessage(
six.b(
json.dumps(
{"a": self._message_subscribe, "v": instrument_tokens}
)
)
)
for token in instrument_tokens:
self.subscribed_tokens[token] = self.MODE_QUOTE
return True
except Exception as e:
self._close(reason="Error while subscribe: {}".format(str(e)))
raise
def unsubscribe(self, *args, **kwargs):
"""
Unsubscribe the given list of instrument_tokens.
- `instrument_tokens` is list of instrument_tokens to unsubscribe.
"""
if self.with_shared_ticks:
fn = sys._getframe().f_code.co_name
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.shm_send_obj((fn, args, kwargs), self.rpc_shm)
for token in args[0]:
try:
self.subscribed_tokens.remove(token)
except KeyError:
pass
else:
try:
instrument_tokens = args[0]
self.ws.sendMessage(
six.b(
json.dumps(
{"a": self._message_unsubscribe, "v": instrument_tokens}
)
)
)
for token in instrument_tokens:
try:
del self.subscribed_tokens[token]
except KeyError:
pass
return True
except Exception as e:
self._close(reason="Error while unsubscribe: {}".format(str(e)))
raise
def set_mode(self, *args, **kwargs):
"""
Set streaming mode for the given list of tokens.
- `mode` is the mode to set. It can be one of the following class constants:
MODE_LTP, MODE_QUOTE, or MODE_FULL.
- `instrument_tokens` is list of instrument tokens on which the mode should be applied
"""
if self.with_shared_ticks:
fn = sys._getframe().f_code.co_name
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.shm_send_obj((fn, args, kwargs), self.rpc_shm)
else:
try:
mode, instrument_tokens = args[:2]
self.ws.sendMessage(
six.b(
json.dumps(
{"a": self._message_setmode, "v": [mode, instrument_tokens]}
)
)
)
# Update modes
for token in instrument_tokens:
self.subscribed_tokens[token] = mode
return True
except Exception as e:
self._close(reason="Error while setting mode: {}".format(str(e)))
raise
def process_subscribed_ticks_over_shm(self) -> None:
if not hasattr(self, "ticks_shm"):
try:
self.ticks_shm = KiteExtUtils.read_shm(name="KiteExtTicker")
except FileNotFoundError:
pass
else:
try:
latest_ticks = KiteExtUtils.shm_recv_obj(self.ticks_shm)
except IndexError:
pass
else:
if isinstance(latest_ticks, dict):
self.ticks.update(
{
k: v
for k, v in latest_ticks.items()
if k in self.subscribed_tokens or k == "order_update"
}
)
del latest_ticks
if not hasattr(self, "rpc_shm"):
try:
self.rpc_shm = KiteExtUtils.read_shm(name="KiteExtTickerRpc")
except FileNotFoundError:
pass
def subscribe_ticks_over_shm(self) -> None:
self.subscribe_ticks = CustomTimer(
"subscribe_tickers_timer",
0.15,
self.process_subscribed_ticks_over_shm,
)
self.subscribe_ticks.start()
def __init__(
self,
user_id: str = "",
enctoken: str = "",
root_uri: str = "wss://ws.zerodha.com",
api_key: str = "kitefront",
user_agent: str = "kite3-web",
version: str = "3.0.9",
share_ticks: bool = False,
with_shared_ticks_only: bool = False,
**kwargs,
) -> None:
self.share_ticks = share_ticks
self.with_shared_ticks = with_shared_ticks_only
if not self.with_shared_ticks:
if user_id == "" or enctoken == "":
raise ValueError(
"`user_id` and `enctoken` value can not be empty, Please Try again with correct value of both `user_id` and `enctoken`."
)
access_token = enctoken
KiteTicker.__init__(self, api_key, access_token, **kwargs)
self.socket_url = (
"{root}?api_key={api_key}"
"&user_id={user_id}"
"&enctoken={enctoken}"
"&user-agent={user_agent}"
"&version={version}".format(
root=root_uri,
api_key=api_key,
user_id=user_id,
enctoken=urllib.parse.quote(enctoken),
user_agent=user_agent,
version=version,
)
)
self.default_divisor = 100.0
self.cds_divisor = 10000000.0
self.bcd_divisor = 10000.0
if self.share_ticks or self.with_shared_ticks:
self.ticks: Dict[str, Any] = {"order_update": {}}
if self.share_ticks:
self.publish_ticks_over_shm()
self.initialize_fetch_rpc_calls()
if self.with_shared_ticks:
self.subscribe_ticks_over_shm()
def _parse_text_message(self, payload):
"""Parse text message."""
# Decode unicode data
if not six.PY2 and type(payload) == bytes:
payload = payload.decode("utf-8")
try:
data = json.loads(payload)
except ValueError:
return
# Order update callback
if self.on_order_update and data.get("type") == "order":
if self.share_ticks:
self.ticks["order_update"] = data
self.on_order_update(self, data)
# Custom error with websocket error code 0
if data.get("type") == "error":
self._on_error(self, 0, data)
def _decode_and_process_binary_packet(
self, packet: Any
) -> Optional[Dict[str, Any]]:
tick: Optional[Dict[str, Any]] = None
instrument_token = self._unpack_int(packet, 0, 4)
# Retrive segment constant from instrument_token
segment = instrument_token & 0xFF
# Add price divisor based on segment
if segment == self.EXCHANGE_MAP["cds"]:
divisor = self.cds_divisor
elif segment == self.EXCHANGE_MAP["bcd"]:
divisor = self.bcd_divisor
else:
divisor = self.default_divisor
# All indices are not tradable
tradable = segment != self.EXCHANGE_MAP["indices"]
# LTP packets
if len(packet) == 8:
tick = {
"tradable": tradable,
"mode": self.MODE_LTP,
"instrument_token": instrument_token,
"last_price": self._unpack_int(packet, 4, 8) / divisor,
}
# Indices quote and full mode
elif len(packet) == 28 or len(packet) == 32:
mode = self.MODE_QUOTE if len(packet) == 28 else self.MODE_FULL
tick = {
"tradable": tradable,
"mode": mode,
"instrument_token": instrument_token,
"last_price": self._unpack_int(packet, 4, 8) / divisor,
"ohlc": {
"high": self._unpack_int(packet, 8, 12) / divisor,
"low": self._unpack_int(packet, 12, 16) / divisor,
"open": self._unpack_int(packet, 16, 20) / divisor,
"close": self._unpack_int(packet, 20, 24) / divisor,
},
}
# Compute the change price using close price and last price
tick["change"] = 0
if tick["ohlc"]["close"] != 0:
tick["change"] = (
(tick["last_price"] - tick["ohlc"]["close"])
* 100
/ tick["ohlc"]["close"]
)
# Full mode with timestamp
if len(packet) == 32:
try:
timestamp = datetime.fromtimestamp(self._unpack_int(packet, 28, 32))
except Exception:
timestamp = None
tick["exchange_timestamp"] = timestamp
# Quote and full mode
elif len(packet) == 44 or len(packet) == 184:
mode = self.MODE_QUOTE if len(packet) == 44 else self.MODE_FULL
tick = {
"tradable": tradable,
"mode": mode,
"instrument_token": instrument_token,
"last_price": self._unpack_int(packet, 4, 8) / divisor,
"last_traded_quantity": self._unpack_int(packet, 8, 12),
"average_traded_price": self._unpack_int(packet, 12, 16) / divisor,
"volume_traded": self._unpack_int(packet, 16, 20),
"total_buy_quantity": self._unpack_int(packet, 20, 24),
"total_sell_quantity": self._unpack_int(packet, 24, 28),
"ohlc": {
"open": self._unpack_int(packet, 28, 32) / divisor,
"high": self._unpack_int(packet, 32, 36) / divisor,
"low": self._unpack_int(packet, 36, 40) / divisor,
"close": self._unpack_int(packet, 40, 44) / divisor,
},
}
# Compute the change price using close price and last price
tick["change"] = 0
if tick["ohlc"]["close"] != 0:
tick["change"] = (
(tick["last_price"] - tick["ohlc"]["close"])
* 100
/ tick["ohlc"]["close"]
)
# Parse full mode
if len(packet) == 184:
try:
last_trade_time = datetime.fromtimestamp(
self._unpack_int(packet, 44, 48)
)
except Exception:
last_trade_time = None
try:
timestamp = datetime.fromtimestamp(self._unpack_int(packet, 60, 64))
except Exception:
timestamp = None
tick["last_trade_time"] = last_trade_time
tick["oi"] = self._unpack_int(packet, 48, 52)
tick["oi_day_high"] = self._unpack_int(packet, 52, 56)
tick["oi_day_low"] = self._unpack_int(packet, 56, 60)
tick["exchange_timestamp"] = timestamp
# Market depth entries.
depth = {"buy": [], "sell": []}
# Compile the market depth lists.
[
depth["sell" if i >= 5 else "buy"].append(
{
"quantity": self._unpack_int(packet, p, p + 4),
"price": self._unpack_int(packet, p + 4, p + 8) / divisor,
"orders": self._unpack_int(
packet, p + 8, p + 10, byte_format="H"
),
}
)
for i, p in enumerate(range(64, len(packet), 12))
]
tick["depth"] = depth
else:
pass
if self.share_ticks:
self.ticks[instrument_token] = tick
return tick
def _parse_binary(self, bin):
"""Parse binary data to a (list of) ticks structure."""
return [
self._decode_and_process_binary_packet(packet)
for packet in self._split_packets(bin) # noqa E501
]
class KiteExt(KiteConnect):
@staticmethod
async def on_request_start(
session: aiohttp.ClientSession,
trace_config_ctx: SimpleNamespace,
params: aiohttp.TraceRequestStartParams,
) -> None:
trace_config_ctx.start = asyncio.get_event_loop().time()
log.debug(
" Request Hook: Initiated HTTP.%s Request To Url: %s | With Header: %s |",
params.method,
params.url,
json.dumps(dict(params.headers)),
) # noqa E501
print("\n")
@staticmethod
async def on_request_end(
session: aiohttp.ClientSession,
trace_config_ctx: SimpleNamespace,
params: aiohttp.TraceRequestEndParams,
) -> None:
elapsed = asyncio.get_event_loop().time() - trace_config_ctx.start
elapsed_msg = f"{elapsed:.3f} Seconds"
text = await params.response.text()
if len(text) > 500:
text = f"{text[:500]}...Truncated to 500 Characters."
log.debug(
" Response Hook: The HTTP.%s Request To Url: %s | Completed In %s | Response Status: %s %s | Response Header: %s | Response Content: %s |",
params.method,
params.url,
elapsed_msg,
params.response.status,
params.response.reason,
json.dumps(dict(params.headers)),
text,
)
print("\n")
@staticmethod
async def on_request_exception(
session: aiohttp.ClientSession,
trace_config_ctx: SimpleNamespace,
params: aiohttp.TraceRequestExceptionParams,
) -> None:
log.debug(
" Request Exception Hook: The HTTP.%s Request To Url: %s | Request Header: %s | Failed With Exception: %s |",
params.method,
params.url,
json.dumps(dict(params.headers)),
str(params.exception),
)
print("\n")
@staticmethod
async def generate_async_client_session(
base_url: str,
connector: aiohttp.TCPConnector,
headers: Dict[str, Union[str, Any]],
timeout: aiohttp.ClientTimeout,
raise_for_status: bool,
trust_env: bool,
trace_configs: Optional[List[aiohttp.TraceConfig]] = None,
cookie_jar: Optional[aiohttp.CookieJar] = None,
) -> aiohttp.ClientSession:
return aiohttp.ClientSession(
base_url=base_url,
connector=connector,
headers=headers,
timeout=timeout,
cookie_jar=cookie_jar,
raise_for_status=raise_for_status,
trust_env=trust_env,
trace_configs=trace_configs,
)
@staticmethod
def totp(qrcode: str) -> str:
return pyotp.TOTP(qrcode).now().zfill(6)
@staticmethod
def start_background_loop(
loop: asyncio.AbstractEventLoop,
) -> Optional[NoReturn]:
asyncio.set_event_loop(loop)
try:
loop.run_forever()
except (KeyboardInterrupt, SystemExit, ProgramKilled):
loop.run_until_complete(loop.shutdown_asyncgens())
if loop.is_running():
loop.stop()
if not loop.is_closed():
loop.close()
def graceful_exit(self) -> None:
if self.with_shared_tick_data:
self.subscribe_ticks.cancel()
if hasattr(self, "ticks_shm") and self.ticks_shm is not None:
KiteExtUtils.close_shm(self.ticks_shm)
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.close_shm(self.rpc_shm)
if self.share_historical_data:
self.publish_historical.cancel()
if self.with_shared_historical_data:
self.subscribe_historical.cancel()
if hasattr(self, "historical_shm") and self.historical_shm is not None:
KiteExtUtils.close_shm(
self.historical_shm,
unlink=self.share_historical_data,
)
with contextlib.suppress(RuntimeError, RuntimeWarning):
asyncio.run_coroutine_threadsafe(self.__reqsession.close(), self.loop)
asyncio.run_coroutine_threadsafe(asyncio.sleep(0.25), self.loop)
asyncio.run_coroutine_threadsafe(self.__web_reqsession.close(), self.loop)
asyncio.run_coroutine_threadsafe(asyncio.sleep(0.25), self.loop)
asyncio.run_coroutine_threadsafe(self.loop.shutdown_asyncgens(), self.loop)
if self.loop.is_running():
self.loop.stop()
if not self.loop.is_closed():
self.loop.close()
def __aenter__(self) -> "KiteExt":
return self
def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
self.graceful_exit()
def __enter__(self) -> "KiteExt":
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.graceful_exit()
def __del__(self) -> None:
self.graceful_exit()
def __delete__(self) -> None:
self.graceful_exit()
def subscribe(self, *args, **kwargs):
fn = sys._getframe().f_code.co_name
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.shm_send_obj((fn, args, kwargs), self.rpc_shm)
self.subscribed_tokens.update(set(args[0]))
def unsubscribe(self, *args, **kwargs):
fn = sys._getframe().f_code.co_name
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.shm_send_obj((fn, args, kwargs), self.rpc_shm)
for token in args[0]:
try:
self.subscribed_tokens.remove(token)
except KeyError:
pass
def set_mode(self, *args, **kwargs):
fn = sys._getframe().f_code.co_name
if hasattr(self, "rpc_shm") and self.rpc_shm is not None:
KiteExtUtils.shm_send_obj((fn, args, kwargs), self.rpc_shm)
def process_subscribed_ticks_over_shm(self) -> None:
if not hasattr(self, "ticks_shm"):
try:
self.ticks_shm = KiteExtUtils.read_shm(name="KiteExtTicker")
except FileNotFoundError:
pass
else:
try:
latest_ticks = KiteExtUtils.shm_recv_obj(self.ticks_shm)
except IndexError:
pass
else:
if isinstance(latest_ticks, dict):
self.ticks.update(
{
k: v
for k, v in latest_ticks.items()
if k in self.subscribed_tokens or k == "order_update"
}
)
del latest_ticks
if not hasattr(self, "rpc_shm"):
try:
self.rpc_shm = KiteExtUtils.read_shm(name="KiteExtTickerRpc")
except FileNotFoundError:
pass
def subscribe_ticks_over_shm(self) -> None:
self.subscribe_ticks = CustomTimer(
"subscribe_tickers_timer",
0.15,
self.process_subscribed_ticks_over_shm,
)
self.subscribe_ticks.start()
def __share_historical_data(
self, list_of_kwargs: List[Dict[str, Any]], return_df: bool = True
) -> None:
historical_datas = self.multiple_historical_data(
list_of_kwargs, return_df=return_df, return_token_wise_df=True
)
if (
historical_datas is not None
and isinstance(historical_datas, dict)
and self.shared_historical_datas is not None
):
self.shared_historical_datas.update(historical_datas)
try:
KiteExtUtils.shm_send_obj(
self.shared_historical_datas, self.historical_shm
)
except FileNotFoundError:
pass
def _share_historical_data(
self,
instrument_tokens: List[int],
refresh_interval: float,
interval: str = "minute",
continuous: bool = False,
oi: bool = False,
from_date_time: Optional[Union[str, datetime]] = None,
to_date_time: Optional[Union[str, datetime]] = None,
time_delta: Optional[timedelta] = None,
no_of_days: Optional[int] = None,
show_as_df: bool = True,
) -> None:
if len(self.historical_instrument_tokens) != 0:
raise ValueError("List Of Instrument Tokens Found To Be A Non Empty Set.")
self.historical_instrument_tokens.update(set(instrument_tokens))
self.shared_historical_datas = dict.fromkeys(self.historical_instrument_tokens)
if from_date_time is None and time_delta is None and no_of_days is None:
from_date_time = datetime.today().replace(hour=9, minute=15, second=0)
if to_date_time is None and time_delta is None and no_of_days is None:
to_date_time = datetime.today().replace(hour=15, minute=30, second=0)
if time_delta is not None and to_date_time is None and from_date_time is None:
to_date_time = datetime.today().replace(hour=15, minute=30, second=0)
from_date_time = to_date_time - time_delta
if no_of_days is not None and to_date_time is None and from_date_time is None:
to_date_time = datetime.today().replace(hour=15, minute=30, second=0)
from_date_time = to_date_time - timedelta(
days=no_of_days, hours=6, minutes=15
)
print(from_date_time, to_date_time)
list_of_kwargs = [
{
"instrument_token": inst,
"from_date": from_date_time,
"to_date": to_date_time,
"interval": interval,
"continuous": continuous,
"oi": oi,
}
for inst in instrument_tokens
]
self.historical_shm = KiteExtUtils.create_shm(name="KiteExtHistorical")
self.publish_historical = CustomTimer(
"historical_data_timer",
refresh_interval - 0.05,
self.__share_historical_data,
(list_of_kwargs,),
{"return_df": show_as_df},
)
self.publish_historical.start()
def __refresh_shared_historical_data(self) -> None:
if not hasattr(self, "historical_shm"):
try:
self.historical_shm = KiteExtUtils.read_shm(name="KiteExtHistorical")
except FileNotFoundError:
pass
try:
latest_historical = KiteExtUtils.shm_recv_obj(self.historical_shm)
except IndexError:
pass
else:
if (
latest_historical is not None
and isinstance(latest_historical, dict)
and self.shared_historical_datas is not None
):
self.shared_historical_datas.update(
{
k: v
for k, v in latest_historical.items()
if k in self.historical_instrument_tokens
}
)
del latest_historical
def refresh_shared_historical_data(
self,
instrument_tokens: Optional[List[int]] = None,
) -> None:
if instrument_tokens is not None:
self.historical_instrument_tokens.update(set(instrument_tokens))
if len(self.historical_instrument_tokens) == 0:
raise ValueError(
"List Of Instrument Tokens, Needed To Synchronize Shared Historical Data Between Sperate Instances / Processes."
)
self.shared_historical_datas = dict.fromkeys(self.historical_instrument_tokens)
self.subscribe_historical = CustomTimer(
"subscribe_tickers_timer",
0.5,
self.__refresh_shared_historical_data,
)
self.subscribe_historical.start()
def __init__(
self,
api_key: str = "kitefront",
user_id: Optional[str] = None,
with_shared_tick_data: bool = False,
share_historical_data: bool = False,
with_shared_historical_data: bool = False,
with_shared_historical_data_only: bool = False,
subscribe_tick_inst_tokens: Optional[List[int]] = None,
subscribe_historical_inst_tokens: Optional[List[int]] = None,
*args,
**kwargs,
):
self.__get_delete_method = {"GET", "DELETE"}
self.__post_put_method = {"POST", "PUT"}
self.ticks: Dict[str, Any] = {"order_update": {}}
self.with_shared_tick_data = with_shared_tick_data
self.subscribed_tokens: Set[int] = (
set(subscribe_tick_inst_tokens)
if subscribe_tick_inst_tokens is not None and with_shared_tick_data
else set()
)
self.share_historical_data = share_historical_data
self.with_shared_historical_data = with_shared_historical_data
self.with_shared_historical_data_only = with_shared_historical_data_only
self.historical_instrument_tokens: Set[int] = (
set(subscribe_historical_inst_tokens)
if subscribe_historical_inst_tokens is not None
and with_shared_historical_data
else set()
)
self.shared_historical_datas: Optional[
Dict[int, Union[Dict[str, Any], pd.DataFrame, None]]
] = None
if not self.with_shared_historical_data_only:
KiteConnect.__init__(self, api_key=api_key, *args, **kwargs)
if user_id is not None:
self.user_id = user_id
self._routes.update(
{
"api.login": "/api/login",
"api.twofa": "/api/twofa",
"api.misdata": "/margins/equity",
"web.historical": "/oms/instruments/historical/{instrument_token}/{interval}",
}
)
self._kite_web_uri = self._default_login_uri[
: self._default_login_uri.find("/", 8)
]
self.__initialize_loop()
self.__user_agent = self._user_agent()
self.__web_user_agent = self._browser_user_agent()
self.__headers = {
"X-Kite-Version": self.kite_header_version,
"User-Agent": self.__user_agent,
}
self.__web_headers = {
"X-Kite-Version": self.kite_header_version + ".0.9",
"User-Agent": self.__web_user_agent,
}
self._initialize_session_params()
if self.with_shared_tick_data:
self.subscribe_ticks_over_shm()
if self.with_shared_historical_data:
self.refresh_shared_historical_data()
def handle_stop_signals(self, *args, **kwargs):
try:
self.graceful_exit()
except Exception as err:
log.error(str(err))
else:
raise SystemExit
def __initialize_loop(self) -> None:
global resolver
if platform.system().lower().find("win") == -1:
self.loop = uvloop.new_event_loop()
else:
self.loop = asyncio.new_event_loop()
if platform.system().lower().startswith("win"):
with contextlib.suppress(ValueError):
for sig in (SIGABRT, SIGINT, SIGTERM):
signal.signal(sig, self.handle_stop_signals)
else:
with contextlib.suppress(ValueError):
for sig in (SIGABRT, SIGINT, SIGTERM, SIGHUP):
self.loop.add_signal_handler(
sig, self.handle_stop_signals
) # noqa E501
self.event_thread = Thread(
target=self.start_background_loop,
args=(self.loop,),
name=f"{self.__class__.__name__}_event_thread",
daemon=True,
)
self.event_thread.start()
log.debug("Asyncio Event Loop has been initialized.")
async def __initialize_session_params(self) -> None:
self.__timeout = aiohttp.ClientTimeout(
total=float(self._default_timeout)
) # noqa E501
self.__resolver = AsyncResolver(nameservers=["1.1.1.1", "1.0.0.1"])
self.__connector = aiohttp.TCPConnector(
limit=100,
limit_per_host=0,
ttl_dns_cache=22500,
use_dns_cache=True,
resolver=self.__resolver,
family=socket.AF_INET,
)
cookie_jar = aiohttp.CookieJar()
self.__trace_config = aiohttp.TraceConfig()
self.__trace_config.on_request_start.append(self.on_request_start)
self.__trace_config.on_request_end.append(self.on_request_end)
self.__trace_config.on_request_exception.append(self.on_request_exception)
self.__reqsession = await self.generate_async_client_session(
base_url=self.root,
connector=self.__connector,
headers=self.__headers,
timeout=self.__timeout,
raise_for_status=True,
trust_env=True,
trace_configs=[self.__trace_config] if self.debug else None,
cookie_jar=cookie_jar,
)
self.__web_reqsession = await self.generate_async_client_session(
base_url=self._default_login_uri[:-14],
connector=self.__connector,
headers=self.__web_headers,
timeout=self.__timeout,
raise_for_status=True,
trust_env=True,
trace_configs=[self.__trace_config] if self.debug else None,
)
await self.__reqsession.get("/")
await self.__web_reqsession.get("/")
async def __fetch_browser_user_agent(self) -> Optional[str]:
async with aiohttp.ClientSession(
base_url="https://techfanetechnologies.github.io",
timeout=aiohttp.ClientTimeout(total=60), # noqa E501
raise_for_status=True,
) as session:
for retry in range(3):
try:
async with session.get(
"/latest-user-agent/user_agents.json"
) as response: # noqa E501
_user_agent = await response.json()
except aiohttp.ClientError as err:
log.error(f"User Agent Fetch Failed with exception {err}")
await asyncio.sleep(1.0)
log.error(f"Retrying..., Retry No.{retry+1}")
else:
return _user_agent[0]
def _initialize_session_params(self) -> None:
future = asyncio.run_coroutine_threadsafe(
self.__initialize_session_params(),
self.loop,
)
try:
result = future.result(float(self._default_timeout))
except TimeoutError:
error_message = f"The Initialization of Async Client Session Took Longer Than The Default Timeout To Wait For The Response, i.e. {float(self._default_timeout):.2f} Seconds, Cancelling The Task..."
log.error(error_message)
future.cancel()
except Exception as exc:
log.exception(
f"The Initialization of Async Client Session Ended Up With An Exception: {exc!r} {future.exception(1.0)}"
)
def login_using_enctoken(
self, userid: str, enctoken: str, public_token: Optional[str] = None
) -> None:
self.user_id = userid
self.enctoken = enctoken
self.public_token = public_token
self.set_headers(enctoken, userid=userid)
async def __login_with_credentials(
self,
userid: str,
password: str,
secret: str,
) -> Optional[Tuple[Optional[str], Optional[str], Optional[str]]]:
try:
async with self.__reqsession.post(
url=self._routes["api.login"],
data={"user_id": self.user_id, "password": self.password},
) as resp:
resp_text = await resp.text()
if not self.debug:
print(
f"The HTTP.{resp.method} Request To Url: {resp.url} | Response Status: {resp.status} {resp.reason} | Response Header: {json.dumps(dict(resp.headers))} | Response Content: {resp_text} |"
)
except aiohttp.ClientError as err:
print(f"The Login Request Failed With An Exception: {err}")
else:
try:
json_body = await resp.json()
except (JSONDecodeError, ValueError):
print(f"The Login Request Has Got A Non-Json Response: { resp_text}")
else:
try:
async with self.__reqsession.post(
url=self._routes["api.twofa"],
data={
"user_id": json_body["data"]["user_id"],
"request_id": json_body["data"]["request_id"],
"twofa_value": self.totp(self.twofa_secret),
},
) as resp:
resp_text = await resp.text()
if not self.debug:
print(
f"The HTTP.{resp.method} Request To Url: {resp.url} | Response Status: {resp.status} {resp.reason} | Response Header: {resp.headers} | Response Content: {resp_text} |"
)
except aiohttp.ClientError as err:
print(
f"The TwoFa Authentication Request Failed With An Exception: {err}"
)
else:
self.reqsession_set_cookies_value = [
str(resp.cookies[item])
for item in (
"user_id",
"enctoken",
"public_token",
"kf_session",
"_cfuvid",
)
if resp.cookies.get(item) is not None
]
cookies = self.__reqsession.cookie_jar.filter_cookies(
self._default_root_uri
)
for key, cookie in cookies.items():
if cookie.key == "enctoken":
self.enctoken = cookie.value
if cookie.key == "user_id":
self.user_id = cookie.value
if cookie.key == "kf_session":
self.kf_session = cookie.value
if cookie.key == "_cfuvid":
self.cfuvid = cookie.value
if resp.cookies.get("public_token") is not None:
self.public_token = (
str(resp.cookies["public_token"])
.split("public_token=")[1]
.split("; ")[0]
)
if (
self.enctoken is not None
and self.public_token is not None
and self.user_id is not None
):
self.set_headers(
self.enctoken,
public_token=self.public_token,
userid=self.user_id,
)
pprint.pprint(
{
"user_id": self.user_id,
"enctoken": self.enctoken,
"public_token": self.public_token,
"kf_session": self.kf_session,
"_cfuvid": self.cfuvid,
}
)
def login_with_credentials(
self,
userid: str,
password: str,
secret: str,
) -> None:
self.user_id = userid
self.password = password
if len(secret) != 32:
raise ValueError("Incorrect TOTP Base32 Secret Key")
self.twofa_secret = secret
self.remove_auth_headers()
future = asyncio.run_coroutine_threadsafe(
self.__login_with_credentials(userid, password, secret),
self.loop,
)
try:
future.result(float(self._default_timeout * 2))
except TimeoutError:
error_message = f"The Login With Credentials Request's Took Longer Than The Default Timeout To Wait For The Response, i.e. {float(self._default_timeout * 2):.2f} Seconds, Cancelling The Task..."
log.error(error_message)
future.cancel()
except Exception as exc:
log.exception(
f"The Login With Credentials Requests Ended Up With An Exception: {exc!r} {future.exception(1.0)}"
)
def _browser_user_agent(self) -> Optional[str]:
future = asyncio.run_coroutine_threadsafe(
self.__fetch_browser_user_agent(),
self.loop,
)
try:
result = future.result(float(self._default_timeout))
except asyncio.TimeoutError:
error_message = f"The Request Took Longer Than The Default Timeout To Wait For The Response, i.e. {float(self._default_timeout):.2f} Seconds, Cancelling The Task..."
log.error(error_message)
future.cancel()
except Exception as exc:
log.exception(f"The Request Ended Up With An Exception: {exc!r}")
else:
return result
def set_headers(
self,
enctoken: str,
public_token: Optional[str] = None,
userid: Optional[str] = None,
) -> None:
self.enctoken = enctoken
self.public_token = public_token
if userid is not None:
self.user_id = userid
else:
raise ValueError(
f"userid cannot be `None`, either login with credentials first or set userid here"
)
auth_header = {
"Authorization": "enctoken {}".format(enctoken),
}
self.__headers.update(auth_header)
self.__web_headers.update(auth_header)
self.__reqsession.headers.update(auth_header)
self.__web_reqsession.headers.update(auth_header)
def remove_auth_headers(self) -> None:
if "Authorization" in self.__headers:
self.__headers.pop("Authorization", None)
if "Authorization" in self.__reqsession.headers:
self.__reqsession.headers.pop("Authorization", None)
async def __feth_historical_data_from_web(
self,
url: str,
params: Dict[str, Union[List[Any], Any]],
limit: asyncio.Semaphore = asyncio.Semaphore(2),
return_df: bool = False,
) -> Optional[
Union[Dict[str, List[List[Union[str, int, float, datetime]]]], pd.DataFrame]
]:
for no_of_retries in range(0, 10):
if no_of_retries > 0:
msg = f"Retry No: {no_of_retries} for Url: {url}"
log.debug(msg)
del msg
async with limit:
try:
response = await self.__web_reqsession.get(
url=url,
params=params,
verify_ssl=not self.disable_ssl,
allow_redirects=True,
)
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
sleep_for = no_of_retries * 1.137
msg = (
f"Fetch of Url: {url} with Params: {params} Met With "
f"An Exception: {e} | Going to sleep for"
f"{sleep_for} Seconds, And Retry Again..."
)
log.debug(msg)
del msg
await asyncio.sleep(sleep_for)
continue
else:
if response.headers.get("content-type") == "application/json":
try:
json_body = await response.json()
except (JSONDecodeError, ValueError):
content = await response.text()
raise ex.DataException(
"Could not parse the JSON response received from the server: {content}".format(
content=content
)
)
else:
if (
isinstance(json_body, dict)
and json_body.get("error_type") is not None
and json_body.get("status") == "error"
):
# Call session hook if its registered and
# TokenException is raised.
if (
self.session_expiry_hook
and response.status == 403
and json_body["error_type"] == "TokenException"
):
self.session_expiry_hook()
# native Kite errors
exp = getattr(
ex, json_body.get("error_type"), ex.GeneralException
)
raise exp(json_body["message"], code=response.status)
return self._format_historical(
json_body["data"], return_df=return_df
)
else:
content = await response.text()
raise ex.DataException(
"Unknown Content-Type ({content_type}) with response: ({content})".format(
content_type=response.headers.get("content-type"),
content=content,
)
)
async def __feth_Multiple_historical_data_from_web(
self,
url_args_params,
limit,
return_df: bool = False,
return_token_wise_df: bool = False,
) -> Union[
Dict[int, Optional[Union[Dict[str, Any], pd.DataFrame]]],
List[Union[Dict[str, Any], pd.DataFrame, None]],
]:
try:
async with asyncio.TaskGroup() as tg:
tasks = {
inst_token: tg.create_task(
self.__feth_historical_data_from_web(
self._routes["web.historical"].format(**url_args),
params,
limit=limit,
return_df=return_df,
)
)
for inst_token, (url_args, params) in url_args_params
}
if return_token_wise_df:
return {inst_token: task.result() for inst_token, task in tasks.items()}
else:
return [task.result() for task in tasks.values()]
except AttributeError:
inst_tokens = [
inst_token for inst_token, (url_args, params) in url_args_params
]
tasks = await asyncio.gather(
*[
asyncio.ensure_future(
self.__feth_historical_data_from_web(
self._routes["web.historical"].format(**url_args),
params,
limit=limit,
return_df=return_df,
)
)
for inst_token, (url_args, params) in url_args_params
]
)
if return_token_wise_df:
return {
inst_token: task.result()
for inst_token, task in zip(inst_tokens, tasks)
}
else:
return [task.result() for task in tasks]
def historical_data(
self,
instrument_token: Union[int, str],
from_date: Union[datetime, str],
to_date: Union[datetime, str],
interval: str,
continuous: bool = False,
oi: bool = False,
web_api: bool = False,
return_df: bool = False,
return_kwargs: bool = False,
) -> Union[
Optional[
Union[
Dict[str, List[List[Union[str, int, float, datetime]]]], pd.DataFrame
],
],
Tuple[Dict[str, Union[str, int]], Dict[str, Union[str, int]]],
]:
"""
Retrieve historical data (candles) for an instrument.
Although the actual response JSON from the API does not have field
names such has 'open', 'high' etc., this function call structures
the data into an array of objects with field names. For example:
- `instrument_token` is the instrument identifier (retrieved from the instruments()) call.
- `from_date` is the From date (datetime object or string in format of yyyy-mm-dd HH:MM:SS.
- `to_date` is the To date (datetime object or string in format of yyyy-mm-dd HH:MM:SS).
- `interval` is the candle interval (minute, day, 5 minute etc.).
- `continuous` is a boolean flag to get continuous data for futures and options instruments.
- `oi` is a boolean flag to get open interest.
"""
date_string_format = "%Y-%m-%d %H:%M:%S"
from_date_string = (
from_date.strftime(date_string_format)
if isinstance(from_date, datetime)
else from_date
)
to_date_string = (
to_date.strftime(date_string_format)
if isinstance(to_date, datetime)
else to_date
)
url_args = {"instrument_token": instrument_token, "interval": interval}
params = {
"from": from_date_string,
"to": to_date_string,
"interval": interval,
"continuous": 1 if continuous else 0,
"oi": 1 if oi else 0,
}
if return_kwargs:
return url_args, params
if not web_api:
data = self._get(
"market.historical",
url_args=url_args,
params=params,
)
return self._format_historical(data, return_df=return_df)
else:
future = asyncio.run_coroutine_threadsafe(
self.__feth_historical_data_from_web(
self._routes["web.historical"].format(**url_args),
params,
return_df=return_df,
),
self.loop,
)
try:
result = future.result(timeout=float(self._default_timeout * 8))
except asyncio.TimeoutError:
error_message = f"The Request Took Longer Than The Default Timeout To Wait For The Response, i.e. {self._default_timeout} Seconds, Cancelling The Task..."
log.error(error_message)
del error_message
future.cancel()
except Exception as exc:
log.exception(f"The Request Ended Up With An Exception: {exc!r}")
else:
return result
def _format_historical(
self,
data: Dict[str, List[List[Union[str, int, float, datetime]]]],
return_df: bool = False,
) -> Union[pd.DataFrame, Dict[str, List[List[Union[str, int, float, datetime]]]]]:
records = [
{
"date": dateutil.parser.parse(candle[0]),
"open": candle[1],
"high": candle[2],
"low": candle[3],
"close": candle[4],
"volume": candle[5],
}
| {"oi": candle[6]}
if len(candle) == 7
else {}
for candle in data["candles"]
]
return pd.json_normalize(records) if return_df else records
def multiple_historical_data(
self,
list_of_kwargs: List[Dict[str, Any]],
return_df: bool = False,
return_token_wise_df: bool = False,
) -> Optional[
Union[
Dict[int, Union[Dict[str, Any], pd.DataFrame, None]],
List[Union[Dict[str, Any], pd.DataFrame, None]],
]
]:
url_args_params = []
for kwargs in list_of_kwargs:
url_args_params.append(
(
kwargs["instrument_token"],
self.historical_data(
instrument_token=kwargs["instrument_token"],
from_date=kwargs["from_date"],
to_date=kwargs["to_date"],
interval=kwargs["interval"],
continuous=kwargs["continuous"],
oi=kwargs["oi"],
web_api=True,
return_kwargs=True,
),
)
)
limit = asyncio.Semaphore(len(url_args_params))
futures = asyncio.run_coroutine_threadsafe(
self.__feth_Multiple_historical_data_from_web(
url_args_params,
limit,
return_df=return_df,
return_token_wise_df=return_token_wise_df,
),
self.loop,
)
try:
results = futures.result(
timeout=float(self._default_timeout * 8 * len(url_args_params))
)
except asyncio.TimeoutError:
error_message = f"The Requests Took Longer Than The Default Timeout To Wait For The Response, i.e. {self._default_timeout} Seconds, Cancelling The Task..."
log.error(error_message)
del error_message
futures.cancel()
except Exception as exc:
log.exception(f"The Requests Ended Up With An Exception: {exc!r}")
else:
return results
async def __request(
self,
url: str,
method: str,
params: Optional[Dict[str, Union[List[Any], Any]]] = None,
data: Optional[Dict[str, Union[List[Any], Any]]] = None,
json: Optional[Dict[str, Union[List[Any], Any]]] = None,
verify_ssl: bool = True,
allow_redirects: bool = True,
proxy: Optional[str] = None,
limit: asyncio.Semaphore = asyncio.Semaphore(2),
) -> Optional[Union[Dict[str, Union[List[Any], Any]], bytes]]:
async with limit:
try:
response = await self.__reqsession.request(
method=method,
url=url,
json=json,
data=data,
params=params,
verify_ssl=verify_ssl,
allow_redirects=allow_redirects,
proxy=proxy if isinstance(proxy, str) else None,
)
except aiohttp.ClientError as err:
raise err
except asyncio.TimeoutError:
return await self.__request(
url,
method,
json=json,
data=data,
params=params,
verify_ssl=verify_ssl,
allow_redirects=allow_redirects,
proxy=proxy,
limit=limit,
)
else:
if response.headers.get("content-type") == "application/json":
try:
json_body = await response.json()
except (JSONDecodeError, ValueError):
content = await response.text()
raise ex.DataException(
"Could not parse the JSON response received from the server: {content}".format(
content=content
)
)
else:
if (
isinstance(json_body, dict)
and json_body.get("error_type") is not None
and json_body.get("status") == "error"
):
# Call session hook if its registered and
# TokenException is raised.
if (
self.session_expiry_hook
and response.status == 403
and json_body["error_type"] == "TokenException"
):
self.session_expiry_hook()
# native Kite errors
exp = getattr(
ex, json_body.get("error_type"), ex.GeneralException
)
raise exp(json_body["message"], code=response.status)
return json_body["data"]
elif response.headers.get("content-type") == "csv":
return await response.read()
else:
content = await response.text()
raise ex.DataException(
"Unknown Content-Type ({content_type}) with response: ({content})".format(
content_type=response.headers.get("content-type"),
content=content,
)
)
def _request(
self,
route: str,
method: str,
url_args: Optional[Dict[str, Union[List[Any], Any]]] = None,
params: Optional[Dict[str, Union[List[Any], Any]]] = None,
query_params: Optional[Dict[str, Union[List[Any], Any]]] = None,
is_json: Union[bool, List[bool]] = False,
) -> Optional[Union[Dict[str, Union[List[Any], Any]], bytes]]:
"""Make an HTTP request."""
# Form a restful URL
if url_args:
uri = self._routes[route].format(**url_args)
else:
uri = self._routes[route]
if method in self.__get_delete_method:
query_params = params
future = asyncio.run_coroutine_threadsafe(
self.__request(
url=uri,
method=method,
json=params if is_json and method in self.__post_put_method else None,
data=params
if not is_json and method in self.__post_put_method
else None,
params=query_params,
verify_ssl=not self.disable_ssl,
allow_redirects=True,
proxy=self.proxies,
),
self.loop,
)
try:
result = future.result(timeout=float(self._default_timeout))
except asyncio.TimeoutError:
error_message = f"The Request Took Longer Than The Default Timeout To Wait For The Response, i.e. {self._default_timeout} Seconds, Cancelling The Task..."
log.error(error_message)
future.cancel()
except Exception as exc:
log.exception(f"The Request Ended Up With An Exception: {exc!r}")
else:
return result
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment