Untitled

import sys
import traceback

import pandas as pd
import numpy as np
import os

COLS_OUTPUT = ["Scenario Number", "Instrument Label", "Market Label","Maturity Label", "Quotation Mode", "Shift Type", "Shift Value", "Initial Value", "Router"]


class AdaptShocks:

    def __init__(self, in_file, date, out_file):
        self.df = pd.read_csv(in_file, sep=',', low_memory=False, index_col=False)
        self.out_file = out_file
        self.date = date
        self.df_out = pd.DataFrame(columns=COLS_OUTPUT)
        self.result = 'done'
        self.market = pd.read_csv("C:/Users/X660874/OneDrive - Santander Office 365/Documentos/RNIM/cambios FR murex/market.csv", sep=';', low_memory=False, index_col=False)

    def execute(self):
        try:
            if len(self.df) != 0:
                self.transform_file()
                self.write_file()
        except:
            self.result = 'Error in adapt transformation python: {}'.format(traceback.format_exc())

    def transform_file(self):

        self.df = self.df[["YIELD_RETURN", "MUREX_instrument"]]
        self.df['YIELD_RETURN'] = self.df['YIELD_RETURN'].str.split(";")
        self.df = self.df.explode('YIELD_RETURN').reset_index(drop=True)
        self.df['YIELD_RETURN'] = pd.to_numeric(self.df['YIELD_RETURN'], errors='coerce')
        self.df['YIELD_RETURN'] = self.df['YIELD_RETURN'].apply(lambda x: f'{x:.10f}').str.rstrip('0').str.rstrip('.')

        start = 4046
        add = 519
        vectorscenarios = np.arange(start, start+add+1)
        Instruments = self.df["MUREX_instrument"].drop_duplicates().tolist()
        data ={
            "MUREX_instrument": [v1 for v1 in Instruments for v2 in vectorscenarios],
            'Scenario Number': [v2 for v1 in Instruments for v2 in vectorscenarios]
        }
        scenarios = pd.DataFrame(data)

        self.market = self.market[['DISPLAYED', 'MARKET']]
        self.df = pd.merge(self.df, self.market, how='left', left_on=['MUREX_instrument'], right_on=['DISPLAYED']).drop(columns=['DISPLAYED'])

        print(len(scenarios))
        print(len(self.df))

        #self.df = pd.merge(self.df, scenarios, how='left', left_on=['MUREX_instrument'], right_on=['MUREX_instrument'])

        self.df_out['Instrument Label'] = self.df['MUREX_instrument']
        self.df_out['Scenario Number'] = scenarios['Scenario Number']
        self.df_out['Market Label'] = self.df['MARKET']
        self.df_out['Maturity Label'] = ""
        self.df_out['Quotation Mode'] = "YIELD"
        self.df_out['Shift Type'] = "Std"
        self.df_out['Quotation Mode'] =""
        self.df_out['Shift Value'] = self.df['YIELD_RETURN']
        self.df_out['Initial Value'] =""
        self.df_out['Router'] =""



    def write_file(self):
        with open(self.out_file, 'w') as f:
            f.write('|' + '|'.join(self.df_out.columns) + '|\n')
            for _, row in self.df_out.iterrows():
                f.write('|' + '|'.join(map(str, row)) + '|\n')
        #self.df_out.to_csv(self.out_file, sep="|", index=False, header=True, lineterminator='|\n')
        if os.path.exists(self.out_file):
            #os.chmod(self.out_file, 0775)
            self.result = 'done'
        else:
            self.result = 'ERR_PYTHON'

    def get_result(self):
        return self.result


if __name__ == "__main__":
    """Check if an argument is a csv or a directory, and process."""

    pd.options.mode.chained_assignment = None

    #inFile = sys.argv[1]
    inFile = "C:/Users/X660874/OneDrive - Santander Office 365/Documentos/RNIM/cambios FR murex/2024-11-29_all_returns.csv"
    #date = sys.argv[2]
    date = "2024-10-11"
    #outFile = sys.argv[3]
    outFile = "C:/Users/X660874/OneDrive - Santander Office 365/Documentos/RNIM/cambios FR murex/zspread.txt"

    adapt = AdaptShocks(inFile, date, outFile)

    adapt.execute()

    print(adapt.get_result())