How to reuse class from another file

user_5730859 0
Viewed 100

I import technicals.py into bot.py and want to reuse the variable sl and tp from the class instance process_candles.

If a constant number is given to sl and tp in bot.py, the script is able to work. However, the desired result is to get variable sl and tp which is calculated in the class instance process_candles. from technicals.py.

snippet technicals.py as below:

    df['PAIR'] = self.pair
    decision = NONE
    tp = 0
    sl = 0
if c[-2]>o[-2]:
    if ca[-1]>h[-2]+0.0010:
        decision = BUY
        tp = ca[-1]+0.010
        sl = l[-2]-0.010 
elif o[-2]>c[-2]:
    if cb[-1]<l[-2]-0.0010:
        decision = SELL 
        tp = cb[-1]-0.010
        sl = h[-2]+0.010                
else:
    decision = NONE

snippet bot.py

 def process_pairs(self):      
        trades_to_make = []
        for pair in self.trade_pairs:
            if self.timings[pair].ready == True:
                self.log_message(f"Ready to trade {pair}") 
                techs = Technicals(self.settings[pair], self.api, pair, GRANULARITY, log=self.tech_log)
                decision = techs.get_trade_decision(self.timings[pair].last_candle)
                print ("process decision")
                print (decision)
                units = decision * self.settings[pair].units


                #tp = "154"
                #sl = "153"

                if units != 0:
                    trades_to_make.append({'pair': pair, 'units': units,'take_profit':tp, 'stop_loss':sl})

Full script are as below:

technicals.py

import pandas as pd
import numpy as np

from defs import BUY, SELL, NONE

class Technicals():
    
    def __init__(self, settings, api, pair, granularity, log=None):
        self.settings = settings
        self.log = log
        self.api = api
        self.pair = pair
        self.granularity = granularity
    
    def log_message(self, msg):
        if self.log is not None:
            self.log.logger.debug(msg)

    
    def fetch_candles(self, row_count, candle_time):
        status_code, df = self.api.fetch_candles(self.pair, count=row_count, granularity=self.granularity)
        if df is None:
            self.log_message(f"Error fetching candles for pair:{self.pair} {candle_time}, df None")
            return None
        elif df.iloc[-1].time != candle_time:
            self.log_message(f"Error fetching candles for pair:{self.pair} {candle_time} vs {df.iloc[-1].time}")
            return None
        else:
            return df

    
    def process_candles(self, df):
        open = df.mid_o
        o = np.array(open,dtype='float') 
        #print (o)
                
        high = df.mid_h
        h = np.array(high,dtype='float') 
        #print (h)

        low = df.mid_l
        l = np.array(low,dtype='float')
        #print (l)

        close = df.mid_c
        c = np.array(close,dtype='float')
        print (c)

        close_ask = df.ask_c
        ca = np.array(close_ask,dtype='float')
        print (ca)

        close_bid = df.bid_c
        cb = np.array(close_bid,dtype='float')
        print (cb)

        df['PAIR'] = self.pair
        decision = NONE
        tp = 0
        sl = 0
    if c[-2]>o[-2]:
        if ca[-1]>h[-2]+0.0010:
            decision = BUY
            tp = ca[-1]+0.010
            sl = l[-2]-0.010 
    elif o[-2]>c[-2]:
        if cb[-1]<l[-2]-0.0010:
            decision = SELL 
            tp = cb[-1]-0.010
            sl = h[-2]+0.010                
    else:
        decision = NONE        

    log_cols = ['time','volume','PAIR','bid_c','ask_c','mid_o','mid_h','mid_l','mid_c']
    self.log_message(f"Processed_df\n{df[log_cols].tail(3)}")
    self.log_message(f"Trade_decision:{decision}")
    self.log_message("")

    return decision


def get_trade_decision(self, candle_time):

    max_rows = self.settings.long_ma + 2
    self.log_message("")
    self.log_message(f"get_trade_decision() pair:{self.pair} max_rows:{max_rows}")

    df = self.fetch_candles(max_rows, candle_time)
    print ("xxxx")
    print (df)
    if df is not None:
        return self.process_candles(df)
    print("get trade decision")
    print(self.process_candles(df))

    return NONE

bot.py

import pprint
import time

from settings import Settings
from log_wrapper import LogWrapper
from timing import Timing
from oanda_api import OandaAPI
from technicals import Technicals
from defs import NONE, BUY, SELL
from trade_manager import TradeManager

GRANULARITY = "M1"
SLEEP = 10.0

class TradingBot():
    
    def __init__(self):    
        self.log = LogWrapper("Bot")
        self.tech_log = LogWrapper("Technicals")
        self.trade_log = LogWrapper("Trade")
        self.trade_pairs = Settings.get_pairs()
        self.settings = Settings.load_settings()
        self.api = OandaAPI()
        self.trade_manager = TradeManager(self.api, self.settings, self.trade_log)
        self.timings = { p: Timing(self.api.last_complete_candle(p, GRANULARITY)) for p in self.trade_pairs }
        self.log_message(f"Bot started with\n{pprint.pformat(self.settings)}")
        self.log_message(f"Bot Timings\n{pprint.pformat(self.timings)}")
        print (self.api)
        
    def log_message(self, msg):
        self.log.logger.debug(msg)       
    
    def update_timings(self):        
        for pair in self.trade_pairs:
            current = self.api.last_complete_candle(pair, GRANULARITY)
            self.timings[pair].ready = False
            if current > self.timings[pair].last_candle:
                self.timings[pair].ready = True
                self.timings[pair].last_candle = current
                self.log_message(f"{pair} new candle {current}")

    def process_pairs(self):      
        trades_to_make = []
        for pair in self.trade_pairs:
            if self.timings[pair].ready == True:
                self.log_message(f"Ready to trade {pair}") 
                techs = Technicals(self.settings[pair], self.api, pair, GRANULARITY, log=self.tech_log)
                decision = techs.get_trade_decision(self.timings[pair].last_candle)
                print ("process decision")
                print (decision)
                units = decision * self.settings[pair].units


                #tp = "154"
                #sl = "153"

                if units != 0:
                    trades_to_make.append({'pair': pair, 'units': units,'take_profit':tp, 'stop_loss':sl})

        if len(trades_to_make) > 0:
            print("bot")
            print(trades_to_make)
            self.trade_manager.place_trades(trades_to_make)
        
    
    def run(self):
        while True:
            self.update_timings()
            self.process_pairs()
            time.sleep(SLEEP)


if __name__ == "__main__":
    b = TradingBot()
    b.run()

defs.py

API_KEY = "xxxx"
ACCOUNT_ID = "xyz"
OANDA_URL = 'https://api-fxpractice.oanda.com/v3'

SECURE_HEADER = {
    'Authorization': f'Bearer {API_KEY}',
    'Content-Type': 'application/json'
}


BUY = 1
SELL = -1
NONE = 0
python
2 Answers

You can define your tp and sl as class variables of Technicals.

class Technicals(object):
    tp: int = 0
    sl: int = 0

and use them within Technicals as:

    cls.tp = ... # if you are inside class-method
    self.tp = ... # if you are inside instance-method

And in the TradingBot you can then simple import Technicals and use the class-Vars like:

tp = Technicals.tp # you can use the class
tp = techs.tp # or the instance you already have
user_100395170

Instead of just returning the decision, also return the take profit and stop loss values:

return decision, tp, sl

Then you can unpack the tuple in process_pairs:

decision, tp, sl = techs.get_trade_decision(self.timings[pair].last_candle)
user_5732550
Related