Convert transactions in CAS to tradelog for TMOAP tracker

cas2tradelog.py

import argparse
import re
import csv
import decimal
import sys
import logging
from getpass import getpass

import pandas as pd
from decimal import Decimal

from casparser import read_cas_pdf

STAMP_DUTY = Decimal("0.00005")


LOGGER = logging.getLogger(__name__)

# ISIN -> Google Finance ticker mapping
ISIN_MAP = {
    "INF109K012M7": "MUTF_IN:ICIC_PRU_NIFT_6H8YUI",
    "INF179K01WN9": "MUTF_IN:HDFC_INDE_SP_1XXI4DY",
    "INF109K01CQ1": "MUTF_IN:ICIC_PRU_CORP_1787KRV",
    "INF740K01OK1": "MUTF_IN:DSP_TAX_SAVE_9IDIGA",
    "INF740KA1DN4": "MUTF_IN:DSP_ARBI_DIR_UN122",
    "INF109K01Q49": "MUTF_IN:ICIC_PRU_LIQU_SM4L1H",
    "INF179KB1HP9": "MUTF_IN:HDFC_LIQU_DIR_1SY277O",
    "INF740K01QD1": "MUTF_IN:DSP_SMAL_CAP_H3MAPW",
    "INF740K01GK7": "MUTF_IN:DSP_STRA_BD_NEGAH6",
    "INF277K01360": "MUTF_IN:TATA_DYNA_BOND_14C9RKC",
    "INF846K01131": "MUTF_IN:AXIS_LONG_TERM_DI868F",
    "INF251K01894": "MUTF_IN:BNP_PARI_LARG_I63VQB",
    "INF090I01JV2": "MUTF_IN:FRAN_INDI_LIQU_HEMT94",
    "INF090I01569": "MUTF_IN:FRAN_INDI_SMAL_1OPIHC3",
    "INF090I01IQ4": "MUTF_IN:FRAN_INDI_SMAL_S6VOWT",
    "INF903J01173": "MUTF_IN:SUND_MID_CAP_TCD3YE",
    "INF082J01036": "MUTF_IN:QUAN_LT_EQUI_1X8K58Y",
    "INF247L01AG2": "MUTF_IN:MOTI_OSWA_SP_VPLTWL",
    "INF846K01DP8": "MUTF_IN:AXIS_BLUE_FUND_17AQV2D",
    "INF179K01XD8": "MUTF_IN:HDFC_CORP_BOND_10KEXP7",
    "INF109K016E5": "MUTF_IN:ICIC_PRU_ALL_2XLBMC",
    "INF082J01168": "MUTF_IN:QUAN_MULT_ASSE_1Q9MXFC",
    "INF109K013N3": "MUTF_IN:ICIC_PRU_SHOR_1II1866",
    "INF179K01YM7": "MUTF_IN:HDFC_ST_DEBT_TVWBP5",
    "INF082J01382": "MUTF_IN:QUAN_INDI_ESG_3483GU",
    "INF082J01150": "MUTF_IN:QUAN_GOLD_SAVI_BHIAEL",
    "INF082J01127": "MUTF_IN:QUAN_LIQU_DIR_1YNBBGV",
    "INF082J01093": "MUTF_IN:QUAN_EQUI_FOF_HHLT2U",
    "INF082J01416": "MUTF_IN:QUAN_NIFT_50_1R4GRR6",
    "INF082J01432": "MUTF_IN:QUAN_SMAL_CAP_8YXEZ5",
    "INF179KC1BG0": "MUTF_IN:HDFC_BANK_FINA_1T22HMY",
    "INF247L01445": "MUTF_IN:MOTI_OSWA_MIDC_1E5B8T2",
}

AMC_MAP = {
    "AXIS Mutual Fund": "Axis MF",
    "Franklin Templeton Mutual Fund": "Franklin MF",
    "HDFC Mutual Fund": "HDFC MF",
    "ICICI Prudential Mutual Fund": "ICICI Prudential MF",
    "MOTILAL OSWAL MUTUAL FUND": "Motilal Oswal MF",
    "Quantum Mutual Fund": "Quantum MF",
}

NAME_MAP = {
    "Quantum Nifty 50 ETF Fund of Fund - Direct Plan Growth": "INF082J01416",
    "Quantum Nifty 50 ETF Fund of Fund - Direct Plan Growth - ISIN: INF082J01416": "INF082J01416",
    "Quantum Small Cap Fund - Direct Plan Growth - ISIN: INF082J01432": "INF082J01432",
}


def round_up(x, exp=Decimal("1.0")):
    return x.quantize(exp, rounding=decimal.ROUND_HALF_UP)


def amount_before_duty(val, p=STAMP_DUTY):
    val = Decimal(val)
    return round_up(val + val * p)


def stamp_duty(val, p=STAMP_DUTY):
    return round_up(Decimal(val) * p, exp=Decimal("0.01"))


def parse_folios(folios):
    for folio in folios:
        yield from parse_folio(folio)


def parse_folio(folio):
    for scheme in folio["schemes"]:
        yield from parse_scheme(scheme, amc=folio["amc"])


def parse_scheme(scheme, amc="", drop_dupes=True):
    """
    warning: Will probably not work if there are transactions from
     before stamp duty was a thing (July 2020)
    """

    frame = pd.DataFrame.from_records(scheme["transactions"])
    if frame.empty:
        return

    frame["isin"] = scheme["isin"]
    frame["amfi"] = scheme["amfi"]
    frame["name"] = scheme["scheme"]
    frame["amount"] = frame["amount"].apply(Decimal)
    frame.loc[frame["units"].notnull(), "units"] = frame.loc[
        frame["units"].notnull(), "units"
    ].apply(Decimal)
    frame["amc"] = amc

    pre_drop = frame.shape
    if drop_dupes is True:
        frame.drop_duplicates(inplace=True)
        post_drop = frame.shape
        if pre_drop != post_drop:
            LOGGER.warning("Dropped duplicates (%s -> %s)", pre_drop, post_drop)

    # filters
    type_purchase = frame["type"].isin(["PURCHASE_SIP", "PURCHASE"])
    type_stamp_duty = frame["type"].isin(["STAMP_DUTY_TAX"])

    frame["fees"] = frame["amount"].apply(amount_before_duty).apply(stamp_duty)
    # ensure stamp duty calculated correctly: breaks for MF mergers, comment out if needed
    assert frame[type_purchase]["fees"].sum() == frame[type_stamp_duty]["amount"].sum()

    yield from frame[~type_stamp_duty].to_dict(orient="records")


class TxnType:
    BUY = "Buy"
    SELL = "Sell"


ZERO = Decimal("0")


def get_finance_ticker(isin):
    return ISIN_MAP[isin]


def get_investment_account(amc):
    return AMC_MAP[amc]


def cas_to_trade_log(cas_dict, out=sys.stdout):
    """
    warning: Only handles BUY AND SELL type transactions
    """

    # (date, transaction_type, symbol, quantity (units), price/unit, amount_before_fees, fees, fund house)
    csv_writer = csv.writer(out)
    for txn in sorted(parse_folios(cas_dict["folios"]), key=lambda x: x["date"]):
        is_purchase = txn["amount"] > ZERO
        # print("::",txn["isin"] or txn["name"])
        csv_writer.writerow(
            [
                txn["date"],
                TxnType.BUY if is_purchase else TxnType.SELL,
                get_finance_ticker(txn["isin"] or NAME_MAP[txn["name"]]),
                txn["units"] if is_purchase else txn["units"] * Decimal("-1.0"),
                txn["nav"],
                txn["amount"] if is_purchase else txn["amount"] * Decimal("-1.0"),
                txn["fees"] if not pd.isnull(txn["fees"]) and is_purchase else "0",
                get_investment_account(txn["amc"]),
            ]
        )


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("file", help="Path to CAS PDF")
    parser.add_argument("-p", "--password", dest="password", help="Password to CAS PDF")
    args = parser.parse_args()
    cas_dict = read_cas_pdf(
        args.file, args.password or getpass("Password:"), force_pdfminer=True
    )
    cas_to_trade_log(cas_dict)


if __name__ == "__main__":
    main()


CLEAN_MF = re.compile(r"(fund|plan|-|option)", re.IGNORECASE)


def clean_name(name):
    return CLEAN_MF.sub("", name).replace("  ", " ").strip()

requirements.txt

casparser==0.4.7a1
pandas==2.2.2