Add Point and Figure parameter to customize number of boxes needed to reverse the trend

src/mplfinance/_utils.py

@@ -389,7 +389,9 @@ def _valid_pnf_kwargs():
         'box_size'    : { 'Default'     : 'atr',
                           'Validator'   : lambda value: isinstance(value,(float,int)) or value == 'atr' },
         'atr_length'  : { 'Default'     : 14,
-                          'Validator'   : lambda value: isinstance(value,int) or value == 'total' },               
+                          'Validator'   : lambda value: isinstance(value,int) or value == 'total' },
+        'reversal'    : { 'Default'     : 1,
+                          'Validator'   : lambda value: isinstance(value,int) }               
     }
 
     _validate_vkwargs_dict(vkwargs)
@@ -884,10 +886,11 @@ def _construct_pointnfig_collections(dates, highs, lows, volumes, config_pointnf
     first to ensure every time there is a trend change (ex. previous box is
     an X, current brick is a O) we draw one less box to account for the price 
     having to move the previous box's amount before creating a box in the 
-    opposite direction. Next we adjust volume and dates to combine volume into 
-    non 0 box indexes and to only use dates from non 0 box indexes. We then
-    remove all 0s from the boxes array and once again combine adjacent similarly
-    signed differences in boxes.
+    opposite direction. During this same step we also combine like signed elements 
+    and associated volume/date data ignoring any zero values that are created by 
+    subtracting 1 from the box value. Next we recreate the box array utilizing a 
+    rolling_change and volume_cache to store and sum the changes that don't break 
+    the reversal threshold.
 
     Lastly, we enumerate through the boxes to populate the line_seg and circle_patches
     arrays. line_seg holds the / and \ line segments that make up an X and 
@@ -931,6 +934,7 @@ def _construct_pointnfig_collections(dates, highs, lows, volumes, config_pointnf
 
     box_size = pointnfig_params['box_size']
     atr_length = pointnfig_params['atr_length']
+    reversal = pointnfig_params['reversal']
 
     if box_size == 'atr':
         if atr_length == 'total':
@@ -945,6 +949,9 @@ def _construct_pointnfig_collections(dates, highs, lows, volumes, config_pointnf
         elif box_size < lower_limit:
             raise ValueError("Specified box_size may not be smaller than (0.01* the Average True Value of the dataset) which has value: "+ str(lower_limit))
 
+    if reversal < 1 or reversal > 9:
+        raise ValueError("Specified reversal must be an integer in the range [1,9]")
+
     alpha  = marketcolors['alpha']
 
     uc     = mcolors.to_rgba(marketcolors['ohlc'][ 'up' ], alpha)
@@ -974,27 +981,82 @@ def _construct_pointnfig_collections(dates, highs, lows, volumes, config_pointnf
     boxes, indexes = combine_adjacent(boxes)
     new_volumes, new_dates = coalesce_volume_dates(temp_volumes, temp_dates, indexes)
 
-    #subtract 1 from the abs of each diff except the first to account for the first box using the last box in the opposite direction
-    first_elem = boxes[0]
-    boxes = [boxes[i]- int((boxes[i]/abs(boxes[i]))) for i in range(1, len(boxes))]
-    boxes.insert(0, first_elem)
-
-    # adjust volume and dates to make sure volume is combined into non 0 box indexes and only use dates from non 0 box indexes
-    temp_volumes, temp_dates = [], []
-    for i in range(len(boxes)):
-        if boxes[i] == 0:
-            volume_cache += new_volumes[i]
-        else:
+    adjusted_boxes = [boxes[0]]
+    temp_volumes, temp_dates = [new_volumes[0]], [new_dates[0]]
+    volume_cache = 0
+
+    # Clean data to subtract 1 from all box # not including the first boxes element and combine like signed adjacent values (after ignoring zeros)
+    for i in range(1, len(boxes)):
+        adjusted_value = boxes[i]- int((boxes[i]/abs(boxes[i])))
+
+        # not equal to 0 and different signs
+        if adjusted_value != 0 and adjusted_boxes[-1]*adjusted_value < 0:
+
+            # Append adjusted_value, volumes, and date to associated lists
+            adjusted_boxes.append(adjusted_value)
             temp_volumes.append(new_volumes[i] + volume_cache)
-            volume_cache = 0
             temp_dates.append(new_dates[i])
-
-    #remove 0s from boxes
-    boxes = list(filter(lambda diff: diff != 0, boxes))
 
-    # combine adjacent similarly signed differences again after 0s removed
-    boxes, indexes = combine_adjacent(boxes)
-    new_volumes, new_dates = coalesce_volume_dates(temp_volumes, temp_dates, indexes)
+            # reset volume_cache once we use it
+            volume_cache = 0
+
+        # not equal to 0 and same signs
+        elif adjusted_value != 0 and adjusted_boxes[-1]*adjusted_value > 0:
+
+            # Add adjusted_value and volume values to last added elements
+            adjusted_boxes[-1] += adjusted_value
+            temp_volumes[-1] += new_volumes[i] + volume_cache
+
+            # reset volume_cache once we use it
+            volume_cache = 0
+
+        else: # adjusted_value == 0
+            volume_cache += new_volumes[i]
+
+    boxes = [adjusted_boxes[0]]
+    new_volumes = [temp_volumes[0]]
+    new_dates = [temp_dates[0]]
+
+    rolling_change = 0
+    volume_cache = 0
+    biggest_difference = 0 # only used for the last column
+
+    #Clean data to account for reversal size (added to allow overriding the default reversal of 1)
+    for i in range(1, len(adjusted_boxes)):
+
+        # Add to rolling_change and volume_cache which stores the box and volume values 
+        rolling_change += adjusted_boxes[i]
+        volume_cache += temp_volumes[i]
+
+        # if rolling_change is the same sign as the previous box and the abs value is bigger than the 
+        # abs value of biggest_difference then we should replace biggest_difference with rolling_change
+        if rolling_change*boxes[-1] > 0 and abs(rolling_change) > abs(biggest_difference):
+            biggest_difference = rolling_change
+
+        # Add to new list if the rolling change is >= the reversal
+        if abs(rolling_change) >= reversal:
+
+            # if rolling_change is the same sign as the previous # of boxes then combine
+            if rolling_change*boxes[-1] > 0:
+                boxes[-1] += rolling_change
+                new_volumes[-1] += volume_cache
+
+            # otherwise add new box
+            else: # < 0 (== 0 can't happen since neither rolling_change or boxes[-1] can be 0)
+                boxes.append(rolling_change)
+                new_volumes.append(volume_cache)
+                new_dates.append(temp_dates[i])
+
+            # reset rolling_change and volume_cache once we've used them
+            rolling_change = 0
+            volume_cache = 0
+
+            # reset biggest_difference as we start from the beginning every time there is a reversal
+            biggest_difference = 0
+
+    # Adjust the last box column if the left over rolling_change is the same sign as the column
+    boxes[-1] += biggest_difference
+    new_volumes[-1] += volume_cache
 
     curr_price = closes[0]
     box_values = [] # y values for the boxes

src/mplfinance/_version.py

@@ -1,5 +1,5 @@
 
-version_info = (0, 12, 7, 'alpha', 14)
+version_info = (0, 12, 7, 'alpha', 15)
 
 _specifier_ = {'alpha': 'a','beta': 'b','candidate': 'rc','final': ''}
 

tests/test_exceptions.py

@@ -81,3 +81,11 @@ def test_figratio_bounds(bolldata):
     with pytest.raises(ValueError) as ex:
         mpf.plot(df,volume=True,figratio=(10,51),savefig=buf)
     assert '"figratio" (aspect ratio)  must be between' in str(ex.value)
+
+def test_reversal_box_size_bounds(bolldata):
+    df = bolldata
+    buf = io.BytesIO()
+    mpf.plot(df,type='pnf',pnf_params=dict(box_size=3, reversal=3), volume=True, savefig=buf)
+    with pytest.raises(ValueError) as ex:
+        mpf.plot(df,type='pnf',pnf_params=dict(box_size=3, reversal=10), volume=True, savefig=buf)
+    assert 'Specified reversal must be an integer in the range [1,9]' in str(ex.value)

tests/test_pnf.py

@@ -109,3 +109,23 @@ def test_pnf04(bolldata):
     if result is not None:
        print('result=',result)
     assert result is None
+
+def test_pnf05(bolldata):
+    df = bolldata
+
+    fname = base+'05.png'
+    tname = os.path.join(tdir,fname)
+    rname = os.path.join(refd,fname)
+
+    mpf.plot(df,type='pnf',pnf_params=dict(box_size='atr',atr_length='total', reversal=2),mav=(4,6,8),volume=True,savefig=tname)
+
+    tsize = os.path.getsize(tname)
+    print(glob.glob(tname),'[',tsize,'bytes',']')
+
+    rsize = os.path.getsize(rname)
+    print(glob.glob(rname),'[',rsize,'bytes',']')
+
+    result = compare_images(rname,tname,tol=IMGCOMP_TOLERANCE)
+    if result is not None:
+       print('result=',result)
+    assert result is None