604 lines
20 KiB
Python
604 lines
20 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Friday, 24th Jan., 2025.
|
|
|
|
OBJECTIVE:
|
|
|
|
To infer EoD data from the ticks collection, and then feed it in the SQL database.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# System-level activities:
|
|
import io
|
|
import os
|
|
|
|
# My utils:
|
|
from utils_v2.string import json
|
|
from utils_v2.string import regex
|
|
from utils_v2.system import files
|
|
from utils_v2.date_time import date_time
|
|
from utils_v2.database.async_mysql_v2 import AsyncMySQL
|
|
from utils_v2.database.async_mongo_v2 import AsyncMongo
|
|
|
|
# To make HTTP calls:
|
|
import httpx
|
|
import socket
|
|
|
|
# To work with date and time:
|
|
import datetime
|
|
import time
|
|
|
|
# To work with datatypes:
|
|
from typing import List, Literal
|
|
|
|
# For handling NaN values:
|
|
import pandas as pd
|
|
|
|
# For scheduling and cron:
|
|
from scheduler.asyncio import Scheduler
|
|
|
|
# For debugging:
|
|
from icecream import IceCreamDebugger
|
|
|
|
# For asynchronous operations:
|
|
import asyncio
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make API calls:
|
|
http_client = httpx.AsyncClient(
|
|
limits = httpx.Limits(
|
|
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
|
|
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
|
|
),
|
|
timeout = httpx.Timeout(
|
|
pool = 120.0, # .... Time to wait for a free connection from the pool.
|
|
connect = 2.5, # ... Time to wait for establishing a connection to the server.
|
|
write = 10.0, # .... Time to wait for sending data.
|
|
read = 120.0 # ..... Time to wait for receiving data.
|
|
)
|
|
)
|
|
|
|
# For debugging:
|
|
printer = IceCreamDebugger(prefix = "EoD (Ticks) | ", includeContext = True)
|
|
no_context_printer = IceCreamDebugger(prefix = "EoD (Ticks) | ", includeContext = False)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# This script's config.:
|
|
SERVER_HOSTNAME = str(socket.gethostname())
|
|
SCRIPT_DATA = {}
|
|
|
|
# For the database:
|
|
sql_writer: AsyncMySQL | None = None
|
|
data_mongo: AsyncMongo | None = None
|
|
|
|
# To construct a report:
|
|
failed_to_parse_dates = []
|
|
failed_eq_dates = []
|
|
failed_fo_dates = []
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
async def init(
|
|
script_id: str,
|
|
debug: bool
|
|
) -> bool:
|
|
|
|
"""
|
|
To initialize all credentials, instances, and connectivity for this whole script.
|
|
:param script_id: The id to use to load cred and data from the internal service.
|
|
:param debug: Whether, or not, you would like to print the debug messages.
|
|
:return: True if initialized successfully, else False.
|
|
"""
|
|
|
|
# Declare the required global variables:
|
|
global SCRIPT_DATA
|
|
global sql_writer
|
|
global data_mongo
|
|
|
|
# Basic stuff:
|
|
if not debug: printer.disable()
|
|
|
|
# ┏┓ ┓ ┓ ┳┓
|
|
# ┃ ┏┓┏┓┏┫ ┏┓┏┓┏┫ ┃┃┏┓╋┏┓
|
|
# ┗┛┛ ┗ ┗┻ ┗┻┛┗┗┻ ┻┛┗┻┗┗┻
|
|
|
|
# Get the script credentials:
|
|
response = await http_client.get(
|
|
url = r"https://nexcom.ditscentre.in/internal/cred/get",
|
|
headers = {"X-Script-Id": script_id}
|
|
)
|
|
if response.status_code not in [200]:
|
|
print("FATAL: SCRIPT CREDENTIALS LOADING FAILED!")
|
|
return False
|
|
script_cred = response.json().get("data")
|
|
|
|
# Get the script data:
|
|
response = await http_client.get(
|
|
url = r"https://nexcom.ditscentre.in/internal/data/get",
|
|
headers = {"X-Script-Id": script_id}
|
|
)
|
|
if response.status_code not in [200]:
|
|
print("FATAL: SCRIPT DATA LOADING FAILED!")
|
|
return False
|
|
SCRIPT_DATA = response.json().get("data")
|
|
|
|
# Done with this step:
|
|
printer("Cred and Data loaded.")
|
|
|
|
# ┳┳┓ • ┳┓┳┓
|
|
# ┃┃┃┏┓┏┓┓┏┓┃┃┣┫
|
|
# ┛ ┗┗┻┛ ┗┗┻┻┛┻┛
|
|
|
|
sql_writer = AsyncMySQL(
|
|
pool_size = script_cred["mariaDb"]["write"]["poolSize"],
|
|
host = script_cred["mariaDb"]["write"]["host"],
|
|
user = script_cred["mariaDb"]["write"]["user"],
|
|
password = script_cred["mariaDb"]["write"]["password"],
|
|
database = script_cred["mariaDb"]["write"]["database"]
|
|
)
|
|
if not await sql_writer.connect():
|
|
print("FATAL: MARIA-DB CONNECTION FAILED!")
|
|
return False
|
|
|
|
printer("MariaDB connected.")
|
|
|
|
# ┳┳┓
|
|
# ┃┃┃┏┓┏┓┏┓┏┓
|
|
# ┛ ┗┗┛┛┗┗┫┗┛
|
|
# ┛
|
|
|
|
data_mongo = AsyncMongo(
|
|
connection_string = script_cred["mongoDb"]["data"]["connectionString"],
|
|
# database_name = script_cred["mongoDb"]["data"]["dbName"],
|
|
database_name = "markets",
|
|
max_connections = script_cred["mongoDb"]["data"]["poolSize"],
|
|
debug = debug
|
|
)
|
|
if not await data_mongo.connect():
|
|
print("FATAL: MONGO-DB NOT CONNECTED!")
|
|
return False
|
|
no_context_printer("MongoDB ready.")
|
|
|
|
# ┳┓
|
|
# ┃┃┏┓┏┓┏┓
|
|
# ┻┛┗┛┛┗┗
|
|
|
|
# If everything went well, we return with success:
|
|
printer("Initialization done.")
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def send_telegram(
|
|
message: str,
|
|
chat_id: str = None,
|
|
message_type: Literal["info", "warning", "error"] = "info"
|
|
) -> None:
|
|
|
|
"""
|
|
To send out alerts and heartbeats to inform about th script being alive.
|
|
:param message: The text to send.
|
|
:param chat_id: The destination chat identifier.
|
|
:param message_type: The kind of message to send. Decides the presentation of the header.
|
|
:return: None.
|
|
"""
|
|
|
|
try:
|
|
|
|
# Create the JSON for sending to the API endpoint:
|
|
json_input = {
|
|
"chatClient": "telegram",
|
|
"message": message,
|
|
"type": message_type
|
|
}
|
|
if chat_id: json_input["chatId"] = json_input
|
|
|
|
# Make the API call to send the ticks:
|
|
response = await http_client.post(
|
|
url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend",
|
|
json = json_input
|
|
)
|
|
|
|
# Raise an exception if the call was not successful:
|
|
response.raise_for_status()
|
|
|
|
# If something goes wrong:
|
|
except Exception as exception:
|
|
printer(message_type, message, exception)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def get_latest_eod_data(target_date: datetime.datetime = None) -> dict:
|
|
|
|
"""
|
|
To fetch the latest EoD (daily candle) data from the ticks database.
|
|
:param target_date: The date (UTC) whose EoD ticks are desired.
|
|
:return: The inferred daily candle data.
|
|
"""
|
|
|
|
no_context_printer("Getting EoD data from ticks.")
|
|
|
|
# Prepare the inputs needed for the aggregation:
|
|
if not isinstance(target_date, datetime.datetime): target_date = date_time.get_current_utc_date_time()
|
|
else: target_date = date_time.to_timezone(target_date, date_time.TIMEZONE_UTC)
|
|
start_ts = target_date.replace(hour = 3, minute = 45, second = 0, microsecond = 0)
|
|
end_ts = target_date.replace(hour = 10, minute = 0, second = 0, microsecond = 0)
|
|
|
|
# Construct the aggregation pipeline:
|
|
# Consider only the target date's ticks:
|
|
stage_0 = {
|
|
"$match": {
|
|
"tradeTs": {
|
|
"$gte": start_ts,
|
|
"$lte": end_ts
|
|
}
|
|
}
|
|
}
|
|
|
|
# Add a field that has the rounded timestamp.
|
|
# We round it to one day for EoD data:
|
|
stage_1 = {
|
|
"$addFields": {
|
|
"roundTs": {
|
|
"$dateTrunc": {
|
|
"date": "$tradeTs",
|
|
"unit": "day",
|
|
"binSize": 1
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
# Now we convert tick to candlesticks:
|
|
stage_2 = {
|
|
"$group": {
|
|
"_id": {
|
|
"roundTs": "$roundTs",
|
|
"symbol": "$symbol"
|
|
},
|
|
"roundTs": {"$last": "$roundTs"},
|
|
"symbol": {"$last": "$symbol"},
|
|
"name": {"$last": "$name"},
|
|
"exchange": {"$last": "$exchange"},
|
|
"segment": {"$last": "$segment"},
|
|
"type": {"$last": "$type"},
|
|
"expiry": {"$last": "$expiry"},
|
|
"strike": {"$last": "$strike"},
|
|
"open": {"$first": "$ltp"},
|
|
"high": {"$max": "$ltp"},
|
|
"low": {"$min": "$ltp"},
|
|
"close": {"$last": "$ltp"},
|
|
"vwap": {"$last": "$vwap"},
|
|
"chg": {"$last": "$chg"},
|
|
"pChg": {"$last": "$pChg"},
|
|
"volume": {"$last": "$totVol"},
|
|
"dayHigh": {"$last": "$h"},
|
|
"dayLow": {"$last": "$l"},
|
|
"ticks": {"$sum": 1},
|
|
"broker": {"$last": "$broker"},
|
|
"brokerToken": {"$last": "$brokerToken"},
|
|
}
|
|
}
|
|
|
|
# Finally we organize and present the data:
|
|
stage_3 = {
|
|
"$sort": {
|
|
"symbol": 1,
|
|
"roundTs": 1
|
|
}
|
|
}
|
|
stage_4 = {
|
|
"$project": {
|
|
"_id": False
|
|
}
|
|
}
|
|
|
|
# Now we run the aggregation:
|
|
eod_data = await data_mongo.aggregate(
|
|
collection = "__cold_zerodhaTicks",
|
|
pipeline = [
|
|
stage_0,
|
|
stage_1,
|
|
stage_2,
|
|
stage_3,
|
|
stage_4
|
|
],
|
|
limit = None,
|
|
raise_exception = False
|
|
)
|
|
|
|
# Done here:
|
|
return eod_data
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def save_latest_eod_data(eod_data: dict) -> bool:
|
|
|
|
"""
|
|
To save the loaded data to the SQL database.
|
|
:param eod_data: The dict of the EoD data received from the ticks database.
|
|
:return: True if successful, else False
|
|
"""
|
|
|
|
# Note down the "scraping time":
|
|
scrape_ts = date_time.get_current_utc_date_time()
|
|
|
|
# Build the query and data content:
|
|
query_str = (
|
|
"INSERT INTO `eod_market_data_today` (exchange, symbol, segment, type, underlying, is_index, expiry, "
|
|
"strike, prev_close, open, high, low, close, ltp, vwap, tot_vol, tot_cash, delivery_vol, delivery_pct, oi, "
|
|
"oi_chg, date, ts, tz, scrape_ts) "
|
|
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);"
|
|
)
|
|
query_data = []
|
|
for data in eod_data:
|
|
total_cash = data["vwap"] * data["volume"]
|
|
target_date = data["roundTs"].strftime("%Y-%m-%d")
|
|
curr_close = data["close"]
|
|
prev_close = curr_close - data["chg"]
|
|
one_query_data = (
|
|
data["exchange"], # ......................... exchange
|
|
data["symbol"], # ........................... symbol
|
|
data["segment"], # .......................... segment
|
|
data["type"], # ............................. type
|
|
data["name"], # ............................. underlying
|
|
None, # ..................................... is_index
|
|
data["expiry"], # ........................... expiry
|
|
data["strike"], # ........................... strike
|
|
prev_close, # ............................... prev_close
|
|
data["open"], # ............................. open
|
|
data["high"], # ............................. high
|
|
data["low"], # .............................. low
|
|
curr_close, # ............................... close
|
|
curr_close, # ............................... ltp
|
|
data["vwap"], # ............................. vwap
|
|
data["volume"], # ........................... tot_vol
|
|
total_cash, # ............................... tot_cash
|
|
None, # ..................................... delivery_vol
|
|
None, # ..................................... delivery_pct
|
|
None, # ..................................... oi
|
|
None, # ..................................... oi_chg
|
|
target_date, # .............................. date
|
|
data["roundTs"].replace(tzinfo = None), # ... ts
|
|
"Asia/Kolkata", # ........................... tz
|
|
scrape_ts.replace(tzinfo = None), # ......... scrape_ts
|
|
)
|
|
one_query_data = [None if pd.isna(d) else d for d in one_query_data]
|
|
query_data.append(one_query_data)
|
|
|
|
# Run the commands:
|
|
no_context_printer("Saving data to SQL DB.")
|
|
rows_affected, db_response, db_exception = await sql_writer.execute_many(
|
|
query = query_str,
|
|
data = query_data,
|
|
return_exception = True
|
|
)
|
|
if db_exception: await send_telegram(
|
|
message = (
|
|
f"*EoD From Ticks:*\n\n"
|
|
"Message: `SQL database threw an exception.`\n\n"
|
|
f"Exception: `{db_exception}`"
|
|
),
|
|
message_type = "error"
|
|
)
|
|
|
|
# Done here:
|
|
success = True if rows_affected else False
|
|
no_context_printer(success)
|
|
return success
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def run_once() -> bool:
|
|
|
|
"""
|
|
Run the job once where data is grabbed from the ticks database and then fed into the SQL database.
|
|
:return: True if successful, else False.
|
|
"""
|
|
|
|
# Get the data from the ticks database:
|
|
eod_data = await get_latest_eod_data()
|
|
if not eod_data:
|
|
await send_telegram(
|
|
message = (
|
|
f"*EoD From Ticks:*\n\n"
|
|
"Message: `Failed to get EoD candles from ticks.`"
|
|
),
|
|
message_type = "error"
|
|
)
|
|
return False
|
|
|
|
# If there is no data to save:
|
|
if not eod_data:
|
|
await send_telegram(
|
|
message = (
|
|
f"*EoD From Ticks:*\n\n"
|
|
"Message: `No EoD data to save to SQL.`"
|
|
),
|
|
message_type = "error"
|
|
)
|
|
return False
|
|
|
|
# Save the data to the SQL database:
|
|
success = await save_latest_eod_data(eod_data)
|
|
if not success:
|
|
await send_telegram(
|
|
message = (
|
|
f"*EoD From Ticks:*\n\n"
|
|
"Message: `Failed to save EoD data to SQL.`"
|
|
),
|
|
message_type = "error"
|
|
)
|
|
return False
|
|
|
|
# If both th steps succeeded, we are good to go:
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def main(
|
|
start_time: datetime.datetime,
|
|
end_time: datetime.datetime,
|
|
interval_seconds: int = 300,
|
|
):
|
|
|
|
"""
|
|
The main scheduler that manages jobs.
|
|
:param start_time: The time of the day at which messages can start going out.
|
|
:param end_time: The time of the day after which new messages should not go out.
|
|
:param interval_seconds: The time (in seconds) between two reminder jobs.
|
|
:return: None.
|
|
"""
|
|
|
|
# Start configuring the scheduler:
|
|
printer("Configuring the schedule-manager.")
|
|
schedule_manager = Scheduler()
|
|
|
|
# Create all the timestamps at which the job must be done:
|
|
all_job_ts = []
|
|
offset_seconds = 0
|
|
while True:
|
|
ts = start_time + datetime.timedelta(seconds = offset_seconds)
|
|
if ts > end_time: break
|
|
all_job_ts.append(ts.time())
|
|
offset_seconds += interval_seconds
|
|
|
|
# Add the jobs:
|
|
for ts in all_job_ts: schedule_manager.daily(ts, run_once)
|
|
printer(len(all_job_ts))
|
|
|
|
# Infinite loop to keep doing the tasks:
|
|
printer("Schedule-manager ready.")
|
|
while True: await asyncio.sleep(3_600)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
# To get args. from the terminal:
|
|
import argparse
|
|
|
|
# Get the config. from the command-line:
|
|
parser = argparse.ArgumentParser(
|
|
description = (
|
|
"To periodically infer EoD data from tick-by-tick data and feed it into the SQL database."
|
|
)
|
|
)
|
|
parser.add_argument(
|
|
"-s", "--script-id",
|
|
dest = "script_id",
|
|
type = str,
|
|
help = "The id of this script (will affect the loaded config)."
|
|
)
|
|
parser.add_argument(
|
|
"--start-time",
|
|
type = str,
|
|
help = "The 24-hr time of the day (in 'HH:MM:SS' format) from which the data can be refreshed."
|
|
)
|
|
parser.add_argument(
|
|
"--end-time",
|
|
type = str,
|
|
help = "The 24-hr time of the day (in 'HH:MM:SS' format) till which the data must be refreshed."
|
|
)
|
|
parser.add_argument(
|
|
"--interval",
|
|
type = int,
|
|
help = "The no. of seconds after which you would like to refresh the data.",
|
|
default = 300
|
|
)
|
|
parser.add_argument(
|
|
"-d", "--debug",
|
|
dest = "debug",
|
|
action = "store_true",
|
|
help = "Whether, or not, you want to see debugging messages in the terminal.",
|
|
default = False
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
|
|
async def runner():
|
|
|
|
# Parse the inputs:
|
|
start_time = datetime.datetime.strptime(args.start_time, "%H:%M:%S")
|
|
end_time = datetime.datetime.strptime(args.end_time, "%H:%M:%S")
|
|
|
|
# Initialize and run the main code:
|
|
if await init(
|
|
script_id = args.script_id,
|
|
debug = args.debug
|
|
): await main(
|
|
start_time = start_time,
|
|
end_time = end_time,
|
|
interval_seconds = args.interval,
|
|
)
|
|
|
|
# Disconnect from the database:
|
|
disconnected = await sql_writer.disconnect()
|
|
# disconnected = await data_mongo.disconnect()
|
|
|
|
|
|
asyncio.run(runner())
|