Run custom Python code while syncing data

Version 5.3.0 introduces a new feature function hooks. It allows you to tap into events while executing init.py and run user-defined Python code. You can use it to run scanners or store the data in alternate locations or whatever your heart desires to do with the data.

To get started:

1. Create an empty Python file with a name of your choice and save it anywhere on your filesystem (even outside the project root).
2. Create or edit src/defs/user.json and add INIT_HOOK with the file path to your Python file.

{
  "INIT_HOOK": "~/Documents/python/eod2_hook.py",
}

In your Python file, we add a function hook called updateNseSymbol. It receives the daily data for Equity stocks. Below, I created a Dictionary containing a watchlist of stock symbols and prices. If the symbols exist in the dictionary and the close price is below the value, it prints an alert.

from datetime import datetime
from typing import Union

alert_dct = {"aplltd": 930.7, "marksans": 152.6}


def updateNseSymbol(
    date: datetime,
    sym: str,
    open: float,
    high: float,
    low: float,
    close: float,
    volume: int,
    total_trades: Union[int, str],
    qty_per_trade: Union[float, str],
    delivery_qty: Union[int, str],
):
    price = alert_dct[sym]

    if sym in alert_dct and close <= price:
        print(f"Alert {date:%d %b %Y}: {sym.upper()} closed below {price}")

When running src/init.py, the alert prints if the condition is satisfied.

You can rewrite the above as a Python Class. The Class must not take any arguments during initialization.

class EOD_hook:
    def __init__(self):
        self.alert_dct = {"aplltd": 930.7, "marksans": 152.6}

    def updateNseSymbol(
        self,
        date: datetime,
        sym: str,
        open: float,
        high: float,
        low: float,
        close: float,
        volume: int,
        total_trades: Union[int, str],
        qty_per_trade: Union[float, str],
        delivery_qty: Union[int, str],
    ):
        price = self.alert_dct[sym]

        if sym in self.alert_dct and close <= price:
            print(f"Alert {date:%d %b %Y}: {sym.upper()} closed below {price}")

If using Python classes, provide the Class name to INIT_HOOK in user.json. It must be separated by a pipe | symbol.

{
  "INIT_HOOK": "~/Documents/python/eod2_hook.py|EOD_hook",
}

Note:

• Type hints are not necessary. They are just added here for clarity.
• Only functions that are defined will be executed. Only include functions you require.

Event/Function name	Description
updateNseSymbol	Receives the Daily price and delivery data for Equity and SME stocks. If delivery data is unavailable, an empty string in passed
updateIndice	Recieves the Daily price data for NSE indices
updatePendingDeliveryData	Receives missing delivery bhavcopy as pandas DataFrame, along with date.
makeAdjustment	Receives the date and a List of Tuples. Each Tuple contains the symbol name and the adjustment factor used to make adjustments.
cleanOutDated	Receives a List of symbols to remove. These are stocks not updated for 365 days.
on_complete	Called when symbols are updated and adjustments completed without errors. If storing data in a Database, use this to commit a transaction or bulk upload data as the case demands.
on_error	Called when an error occurs while updating symbols or making adjustments. If storing data in a Database, use this to abort a transaction or cancel any tasks as the case demands.

Below are the function signatures for the various hooks.

def updatePendingDeliveryData(delivery_bhav: pd.DataFrame, date: datetime):

def updateNseSymbol(
    date: datetime,
    sym: str,
    open: float,
    high: float,
    low: float,
    close: float,
    volume: int,
    total_trades: Union[int, str],
    qty_per_trade: Union[float, str],
    delivery_qty: Union[int, str],
):

def updateIndice(
    date: datetime,
    sym: str,
    open: float,
    high: float,
    low: float,
    close: float,
    volume: int,
):

def makeAdjustment(date: datetime, sym_list: List[Tuple[str, float]]):
    """sym_list is a list of tuples

    Each Tuple contains the symbol name and adjustment factor.

    Formula to apply adjustment

    adjusted_price = price / adjustment_factor

    To Round to nearest 0.05: 

    round(adjusted_price / 0.05) * 0.05
    """

def cleanOutDated(sym_list: List[str]):
    """
    sym_list: A list of symbols to remove
    """

def on_complete():
    """
    Called after all symbols are updated and adjustments complete
    """

def on_error():
    """
    Called if an error occurs
    """