add AreaAware trades

tests/trading_journal/data_for_tests.py

@@ -31,6 +31,14 @@
     ("failure", -53.0)
 ]
 
+# trades outcome used in test_AwareTrade.py
+trades_outcome1 = [
+    ("n.a.", 86), # outcome + pips
+    ("n.a.", 125.9),
+    ("n.a.", 140.5),
+    ("failure", -61.0)
+]
+
 start_hours = [
     (datetime.datetime(2023, 12, 9, 9, 1), datetime.datetime(2023, 12, 9, 9, 0), "H4"),
     (datetime.datetime(2023, 12, 9, 14, 37), datetime.datetime(2023, 12, 9, 13, 0), "H4"),

tests/trading_journal/test_AwareTrade.py

@@ -0,0 +1,80 @@
+import pytest
+
+from trading_journal.open_trade import Trade, AwareTrade
+from forex.candle import Candle
+
+from data_for_tests import (trades1,
+                            trades_entered,
+                            trades_outcome1)
+
+trade_details = ["2020-02-19T21:00:00", "EUR_GBP", "long", "D", 0.83585, 0.82467]
+
+@pytest.mark.parametrize(
+    "TP, RR, exp_instance",
+    [(0.86032, None, AwareTrade), (None, 1.5, AwareTrade), (None, None, Exception)],
+)
+def test_instantiation(TP, RR, exp_instance):
+    if exp_instance is AwareTrade:
+        AwareTrade(
+            RR=RR,
+            TP=TP,
+            start=trade_details[0],
+            pair=trade_details[1],
+            type=trade_details[2],
+            timeframe=trade_details[3],
+            entry=trade_details[4],
+            SL=trade_details[5],
+            init_clist=True
+        )
+    elif exp_instance == Exception:
+        with pytest.raises(Exception):
+            AwareTrade(
+                RR=RR,
+                TP=TP,
+                start=trade_details[0],
+                pair=trade_details[1],
+                type=trade_details[2],
+                timeframe=trade_details[3],
+                entry=trade_details[4],
+                SL=trade_details[5],
+                init_clist=True
+            )
+
+def test_init_clist():
+    """This test checks the init_clist function"""
+    td = AwareTrade(
+        TP=0.86032,
+        start=trade_details[0],
+        pair=trade_details[1],
+        type=trade_details[2],
+        timeframe=trade_details[3],
+        entry=trade_details[4],
+        SL=trade_details[5],
+        init_clist=True,
+    )
+    assert len(td.clist.candles) == 4104
+
+
+trade_data1 = [(*trade, pip) for trade, pip in zip(trades_entered, trades_outcome1)]
+
+@pytest.mark.parametrize("start,type,SR,SL,TP,entry,trades_outcome", trade_data1)
+def test_run(start, type, SR, SL, TP, entry, trades_outcome, clOH8_2019_pickled, clO_pickled):
+    """Test 'run' function"""
+    t = AwareTrade(
+        id="test",
+        start=start,
+        pair="AUD_USD",
+        timeframe="H8",
+        type=type,
+        SR=SR,
+        SL=SL,
+        TP=TP,
+        entry=entry,
+        clist=clOH8_2019_pickled,
+        clist_tm=clO_pickled,
+        connect=False)
+    t.initialise()
+    t.run()
+    assert t.outcome == trades_outcome[0]
+    assert t.pips == trades_outcome[1]
+

tests/trading_journal/test_Trade.py

@@ -65,6 +65,36 @@ def test_get_lasttime_with_pad(start, type, SR, SL, TP, entry, lasttime, clO_pic
 
     assert t.get_lasttime(pad=30) == lasttime
 
+# add a non initialised trade
+trades_for_test_initialise = trades1[:]
+trades_for_test_initialise.append(("2018-12-23 22:00:00", "long", 0.70112, 0.69637, 0.72756, 0.70895))
+entry_data = [(True, "2016-12-29T08:00:00"),
+            (True,"2017-04-12T19:00:00"), 
+            (True,"2017-09-12T01:00:00"),
+            (True, "2018-05-04T01:00:00"),
+            (False, "n.a.")]
+trade_data2 = [(*trade, pip) for trade, pip in zip(trades_for_test_initialise, entry_data)]
+
+@pytest.mark.parametrize("start,type,SR,SL,TP,entry,entry_data", trade_data2)
+def test_initialise(start, type, SR, SL, TP, entry, entry_data, clO_pickled):
+    t = Trade(
+        id="test",
+        start=start,
+        pair="AUD_USD",
+        timeframe="D",
+        type=type,
+        SR=SR,
+        SL=SL,
+        TP=TP,
+        entry=entry,
+        clist=clO_pickled,
+        clist_tm=clO_pickled)
+    t.initialise()
+    assert t.entered == entry_data[0]
+    if hasattr(t, "entry_time"):
+        assert t.entry_time == entry_data[1]
+
+
 trades_for_entry_onrsi = trades1[:]
 trades_for_entry_onrsi.append(("2017-07-25 22:00:00", "short", 0.79743, 0.80577, 0.77479, 0.79343))
 

tests/trading_journal/test_UnawareTrade.py

@@ -2,8 +2,7 @@
 
 from trading_journal.open_trade import UnawareTrade
 
-from data_for_tests import (trades1,
-                            trades_entered,
+from data_for_tests import (trades_entered,
                             trades_outcome)
 
 trade_details = ["2020-02-19T21:00:00", "EUR_GBP", "long", "D", 0.83585, 0.82467]
@@ -53,36 +52,6 @@ def test_init_clist():
     )
     assert len(td.clist.candles) == 4104
 
-# add a non initialised trade
-trades_for_test_initialise = trades1[:]
-trades_for_test_initialise.append(("2018-12-23 22:00:00", "long", 0.70112, 0.69637, 0.72756, 0.70895))
-entry_data = [(True, "2016-12-29T08:00:00"),
-              (True,"2017-04-12T19:00:00"), 
-              (True,"2017-09-12T01:00:00"),
-              (True, "2018-05-04T01:00:00"),
-              (False, "n.a.")]
-trade_data2 = [(*trade, pip) for trade, pip in zip(trades_for_test_initialise, entry_data)]
-
-@pytest.mark.parametrize("start,type,SR,SL,TP,entry,entry_data", trade_data2)
-def test_initialise(start, type, SR, SL, TP, entry, entry_data, clO_pickled):
-    t = UnawareTrade(
-        id="test",
-        start=start,
-        pair="AUD_USD",
-        timeframe="D",
-        type=type,
-        SR=SR,
-        SL=SL,
-        TP=TP,
-        entry=entry,
-        clist=clO_pickled,
-        clist_tm=clO_pickled)
-    t.initialise()
-    assert t.entered == entry_data[0]
-    if hasattr(t, "entry_time"):
-        assert t.entry_time == entry_data[1]
-
-
 trade_data1 = [(*trade, pip) for trade, pip in zip(trades_entered, trades_outcome)]
 
 @pytest.mark.parametrize("start,type,SR,SL,TP,entry,trades_outcome", trade_data1)

trading_journal/open_trade.py

@@ -63,6 +63,21 @@ def append_trademanagement_candles(self,
         if len(self.preceding_candles) > self.candle_number:
             self.preceding_candles = self.preceding_candles[(self.candle_number)*-1:]
 
+    def process_trademanagement(self, d: datetime, fraction: float):
+        """Process trademanagement candles and ajust SL if required"""
+        # align 'd' object to 'trade_params.clisttm_tf' timeframe
+        aligned_d = process_start(dt=d, timeframe=trade_params.clisttm_tf)
+        self.append_trademanagement_candles(aligned_d, fraction)
+
+        if len(self.preceding_candles) == self.candle_number:
+            res = self.check_if_against()
+            if res is True:
+                new_SL = adjust_SL(pair=self.pair, 
+                                    type=self.type,
+                                    list_candles=self.preceding_candles)
+                self.SL.price = new_SL
+            self.preceding_candles = list()
+
     def check_if_against(self):
         """Function to check if middle_point values are
         agaisnt the trade
@@ -155,7 +170,6 @@ def run(self) -> None:
         current_date = datetime.now().date()
         count = 0
         for d in gen_datelist(start=self.start, timeframe=self.timeframe):
-            print(d)
             if d.date() == current_date:
                 logging.warning("Skipping, as unable to end the trade")
                 self.outcome = "future"
@@ -169,18 +183,68 @@ def run(self) -> None:
             if cl is None:
                 count -= 1
                 continue
-            # align 'd' object to 'trade_params.clisttm_tf' timeframe
-            aligned_d = process_start(dt=d, timeframe=trade_params.clisttm_tf)
-            self.append_trademanagement_candles(aligned_d, fraction)
-
-            if len(self.preceding_candles) == self.candle_number:
-                res = self.check_if_against()
-                if res is True:
-                    new_SL = adjust_SL(pair=self.pair, 
-                                       type=self.type,
-                                       list_candles=self.preceding_candles)
-                    self.SL.price = new_SL
+            self.process_trademanagement(d=d, fraction=fraction)
+            self.calculate_overlap(cl=cl)
+            if count >= trade_params.numperiods:
+                self.completed = True
+                t_logger.warning(
+                    "No outcome could be calculated in the "
+                    "trade_params.numperiods interval"
+                )
+                self.outcome = "n.a."
+        self.finalise_trade(cl=cl)
+
+class AwareTrade(OpenTrade):
+    """Class to represent an open Trade of the 'area_aware' type"""
+
+    def __init__(self, **kwargs):
+        """Constructor"""
+        super().__init__(**kwargs)
+
+    def isin_profit(self, price: float) -> bool:
+        """method to calculate if price is in profit.
+        
+        Argument:
+        price: Check if price is in profit area 
+        """
+        if self.type == "long":
+            if price >= self.entry.price:
+                return True
+        if self.type == "short":
+            if price <= self.entry.price:
+                return True
+        return False
+
+
+    def run(self) -> None:
+        """Method to run this AwareTrade.
+
+        This function will run the trade and will set the outcome attribute
+        """
+        fraction = check_timeframes_fractions(timeframe1=self.timeframe,
+                                              timeframe2=trade_params.clisttm_tf)
+
+        current_date = datetime.now().date()
+        count = 0
+        for d in gen_datelist(start=self.start, timeframe=self.timeframe):
+            if d.date() == current_date:
+                logging.warning("Skipping, as unable to end the trade")
+                self.outcome = "future"
+                break
+            if d > self.clist.candles[-1].time and self.connect is False:
+                raise Exception("No candle is available in 'clist' and connect is False. Unable to follow")
+            if self.completed:
+                break
+            count += 1
+            cl = self.fetch_candle(d)
+            if cl is None:
+                count -= 1
+                continue
+            if self.isin_profit(price=cl.c):
+                self.process_trademanagement(d=d, fraction=fraction)
+            else:
                 self.preceding_candles = list()
+
             self.calculate_overlap(cl=cl)
             if count >= trade_params.numperiods:
                 self.completed = True
@@ -190,3 +254,4 @@ def run(self) -> None:
                 )
                 self.outcome = "n.a."
         self.finalise_trade(cl=cl)
+