Here's the hack I implemented for now. It's based on Pyrenko code I found on Github. In my next(), after I pass the data, I access the brick to generate my signal.
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.patches as patches
'''
Formation of New Renko Bricks
During a trend continuation, a new brick of the same color is added to the chart when the Renko Box Size value,
plus one tick, is met. A reversal in the Renko trend and the formation of a different color brick is formed
when price surpasses the Renko Box Size parameter by twice the size plus one tick.
'''
class Renko:
def __init__(self, HLC_history=None, auto=False, brick_size=1.0):
self.source_prices = []
self.renko_prices = []
self.renko_directions = []
if auto:
self.brick_size = self.__get_optimal_brick_size(HLC_history.iloc[:, [0, 1, 2]])
else:
self.brick_size = brick_size
def __renko_rule(self, close):
# Get the gap between two prices
if self.brick_size == 0:
gap_div = 0
else:
gap_div = int(float(close - self.renko_prices[-1]) / self.brick_size)
is_new_brick = False
start_brick = 0
num_new_bars = 0
# When we have some gap in prices
if gap_div != 0:
# Forward any direction (up or down)
if (gap_div > 0 and (self.renko_directions[-1] > 0 or self.renko_directions[-1] == 0)) or (gap_div < 0 and (self.renko_directions[-1] < 0 or self.renko_directions[-1] == 0)):
num_new_bars = gap_div
is_new_brick = True
start_brick = 0
# Backward direction (up -> down or down -> up)
elif np.abs(gap_div) >= 2: # Should be double gap at least
num_new_bars = gap_div
num_new_bars -= np.sign(gap_div)
start_brick = 2
is_new_brick = True
self.renko_prices.append(self.renko_prices[-1] + 2 * self.brick_size * np.sign(gap_div))
self.renko_directions.append(np.sign(gap_div))
#else:
#num_new_bars = 0
if is_new_brick:
# Add each brick
for d in range(start_brick, np.abs(gap_div)):
self.renko_prices.append(self.renko_prices[-1] + self.brick_size * np.sign(gap_div))
self.renko_directions.append(np.sign(gap_div))
return num_new_bars
# Getting renko on history
def build_history(self, prices):
if len(prices) > 0:
# Init by start values
self.source_prices = prices
self.renko_prices.append(prices.iloc[0])
self.renko_directions.append(0)
# For each price in history
for p in self.source_prices[1:]:
self.__renko_rule(p)
return len(self.renko_prices)
# Getting next renko value for last price
def do_next(self, close):
if len(self.renko_prices) == 0:
self.source_prices.append(close)
self.renko_prices.append(close)
self.renko_directions.append(0)
return 1
else:
self.source_prices.append(close)
return self.__renko_rule(close)
# Simple method to get optimal brick size based on ATR
@staticmethod
def __get_optimal_brick_size(HLC_history, atr_timeperiod=14):
brick_size = 0.0
# If we have enough of data
if HLC_history.shape[0] > atr_timeperiod:
brick_size = np.median(talib.ATR(high=np.double(HLC_history.iloc[:, 0]),
low=np.double(HLC_history.iloc[:, 1]),
close=np.double(HLC_history.iloc[:, 2]),
timeperiod=atr_timeperiod)[atr_timeperiod:])
return brick_size
def evaluate(self, method='simple'):
balance = 0
sign_changes = 0
price_ratio = len(self.source_prices) / len(self.renko_prices)
if method == 'simple':
for i in range(2, len(self.renko_directions)):
if self.renko_directions[i] == self.renko_directions[i - 1]:
balance = balance + 1
else:
balance = balance - 2
sign_changes = sign_changes + 1
if sign_changes == 0:
sign_changes = 1
score = balance / sign_changes
if score >= 0 and price_ratio >= 1:
score = np.log(score + 1) * np.log(price_ratio)
else:
score = -1.0
return {'balance': balance, 'sign_changes:': sign_changes,
'price_ratio': price_ratio, 'score': score}
def get_renko_prices(self):
return self.renko_prices
def get_renko_directions(self):
return self.renko_directions
def plot_renko(self, col_up='g', col_down='r'):
fig, ax = plt.subplots(1, figsize=(20, 10))
ax.set_title('Renko chart')
ax.set_xlabel('Renko bars')
ax.set_ylabel('Price')
# Calculate the limits of axes
ax.set_xlim(0.0,
len(self.renko_prices) + 1.0)
ax.set_ylim(np.min(self.renko_prices) - 3.0 * self.brick_size,
np.max(self.renko_prices) + 3.0 * self.brick_size)
# Plot each renko bar
for i in range(1, len(self.renko_prices)):
# Set basic params for patch rectangle
col = col_up if self.renko_directions[i] == 1 else col_down
x = i
y = self.renko_prices[i] - self.brick_size if self.renko_directions[i] == 1 else self.renko_prices[i]
height = self.brick_size
# Draw bar with params
ax.add_patch(
patches.Rectangle(
(x, y), # (x,y)
1.0, # width
self.brick_size, # height
facecolor=col
)
)
plt.show()
