Simple Price Chart Patterns With Python
Simple Price Chart Patterns With Python
The objective in this chapter is to learn how to write scripts capable of recognizing, storing and parameterizing price
chart patterns. This is an exceptionally useful tool for back-testing and running statistical inference on certain
strategies. Even more importantly these patterns will be the basis for training machine learning models.
To define a pattern (sample in ML) we must find start and end points, as indices (numbers of specific rows) of our
data frame, this is will arguably the toughest part. Once we have a start and end point for our sample we can add
and or compute values (features in ML) that characterize it, these values will be grouped together in a list, tuple or
row of an array or data frame, optionally with the start and end indices as well.
We'll then store all of our patterns in data frames or arrays. Notice than in order to do that each sample needs to
have the same length (number of features).
Simple Strategy
Consider a simple strategy, for example buying when two moving averages cross, we have no idea if this will be
profitable or not, but we can find out.
Lets start by defining functions to get data, compute moving averages and find crossovers:
kurl = '[Link]
try:
req = [Link](kurl, params=params, timeout=10)
req.raise_for_status()
except [Link] as e:
print(f"[Error] {symbol} {tf} — {e}")
break
try:
resp = [Link]()
except ValueError:
print(f"[JSON Error] {symbol} {tf} — Invalid response")
break
if not resp:
break
idx = [Link](cross)[0] + 3
return idx
For a generic symbol and timeframe our data will look like:
Now we can pair every start with the closest end by running:
pairs[-5:]
Out[158… array([[2940, 3490],
[3716, 3969],
[4068, 4311],
[4324, 4427],
[4460, 4800]])
We can also check to see if we found what we were looking for by plotting the data:
s = 3700
e = 4000
plots = [
mpf.make_addplot(df['fast'].iloc[s:e], type= 'line', color= 'grey', width= 3, label= 'fast'),
mpf.make_addplot(df['slow'].iloc[s:e], type= 'line', color= 'black', width= 3, label= 'slow')
]
In [160… s = 4300
e = 4450
plots = [
mpf.make_addplot(df['fast'].iloc[s:e], type= 'line', color= 'grey', width= 3, label= 'fast'),
mpf.make_addplot(df['slow'].iloc[s:e], type= 'line', color= 'black', width= 3, label= 'slow')
]
we won't always be able to complete pairs of start and end indices, that's why we have to apply a mask to
pairs and exclude nan that arise from [Link] (using full with [Link] turns the array type to float hence
the astype(int));
when using any indicator that computes values on a certain length of kline it would be best to only consider
indices after the greatest length used.
Features
We now have an array containing our patterns, but we only have two features and to be honest start and end aren't
of much use except for calculating how long the pattern lasts or viewing and checking if the results match our
objective.
There are plenty of additional features we could be considering, arguably the most important is profit we would have
by simply buying when the crossover occurs and selling at the crossunder:
Notice how we now defined a new array, this is because when stacking a column made of float values it will turn all
other values into that type, rendering our first two columns useless as indices for similar computations.
At this point we could try and see how many times we would have made profit and what the average profit for trade
would be:
Other important features could be maximum and minimum points reached, but let's consider something important
before computing them:
as a rule of thumb we should always try to normalize our features so that we can apply the same patterns on
different symbols, for example, and expect values in the same scale and order of magnitude, this aspect will
become increasingly important as we go on.
So that being said we'll compute the maximum and minimum points reached in percentage from our starting point:
The features used mostly depend on the type of strategy we're studying, but they'll likely be subjective as well.
Statistical Inference
Given the nature of our new features, analyzing them like we did with pnl won't really give relevant results.
So it's worth taking some time to learn some basic statistical tools. According to the Central Distribution Theorem,
with enough data we can assume that our features follow a normal distribution (our patterns aren't many so it's a
bit of stretch but we'll try anyway for instructional purposes)
X ∼ N (μ, σ)
μ is the mean and σ is the standard deviation. This assumption is important because it allows us to carry out the
following calculations:
we can calculate the probability P (X > t) , also known as tail probability (survival function), which mean that
by setting a certain threshold t we can find the probability that or feature be greater than it,
t − μ
P (X > t) = 1 − Φ ( )
σ
t = 1
or we can carry out the inverse operation, or evaluate the quantile function, so for a any given probability we
can find the minimum possible value that our features should present,
P (X > t) = p
in python:
In [166… p = 0.3
In [167… p = 0.7
These operations are all we'll need for now, as they're great to help us figure the risk / reward we can set in our
strategy for example.
More Data
If we weren't satisfied with the number of patterns found, we could either attempt to get more klines or, thanks to
the fact that we normalized our features, we could group together patterns from different symbols.
In order to simplify the script, we'll first define a function that finds our patterns:
closes = df['close'].to_numpy()
lgt = pairs[:, 1] - pairs[:, 0]
max_vals = [Link](len(pairs))
min_vals = [Link](len(pairs))
Notice how we dropped the pairs of indices as they lose their significance when comparing patterns on different
symbols, but we did add lgt as a feature.
At this point we iterate over the symbols we want to consider, get the patterns and then group together:
first = 0
for symbol in symbols:
sym = symbol + 'USDT'
print(symbol)
df = kline(sym, tf, 5000)
df['fast'] = ema(df['close'], 50)
df['slow'] = ema(df['close'], 200)
if first == 0:
patts = crossPatterns(df)
print([Link](patts))
first += 1
else:
patts0 = crossPatterns(df)
print([Link](patts0))
patts = [Link]([patts, patts0])
print([Link](patts))
BTC
(12, 4)
ETH
(12, 4)
(24, 4)
SOL
(16, 4)
(40, 4)
Lets now run our statistical inference and see what we get:
In [170… t = 1
p = 0.7
One important thing to note is that we grouped together patterns from different symbols on the same timeframe, i
advise to not proceed the other way around as the results will tend to be on a much different scale, we can check this
by comparing mean values:
for tf in tfs:
print('-'*10)
first = 0
for symbol in symbols:
sym = symbol + 'USDT'
df = kline(sym, tf, 5000)
df['fast'] = ema(df['close'], 50)
df['slow'] = ema(df['close'], 200)
if first == 0:
patts = crossPatterns(df)
print(f'{tf} - {symbol} PnL: {[Link](patts[:, -1])}')
first += 1
else:
patts0 = crossPatterns(df)
print(f'{tf} - {symbol} PnL: {[Link](patts0[:, -1])}')
patts = [Link]([patts, patts0])
----------
1m - BTC PnL: 0.20581467590465724
1m - ETH PnL: 0.02173255555795231
1m - SOL PnL: 0.22944089559209901
All patts PnL: 0.1429513360791419
----------
15m - BTC PnL: -0.3410925410879499
15m - ETH PnL: -0.5522653534508974
15m - SOL PnL: 0.10611273291158346
All patts PnL: -0.29374381444629555
----------
1h - BTC PnL: 1.3945711027269019
1h - ETH PnL: 5.80205572281903
1h - SOL PnL: 1.4869395909221026
All patts PnL: 2.75376388403262
----------
4h - BTC PnL: 13.432611473021455
4h - ETH PnL: 8.388097400073388
4h - SOL PnL: 24.914800166100264
All patts PnL: 15.438377622149813
----------
1d - BTC PnL: 16.747407946906044
1d - ETH PnL: 50.43608178141643
1d - SOL PnL: 741.3685623712988
All patts PnL: 163.22004599373804
Obviously the closer the two timeframes the closer the scale of their features, but it can change drastically for very
different timeframe, while for symbols on the same timeframe scale and order of magnitude stay more or less
consistent.
Other Strategies
This way we have both long and short strategies paired together as the same pattern, but we also have a way of
identifying which type of position we're considering
In [ ]: def crossPatternsLS(df):
closes = df['close'].to_numpy()
lgt = abs(pairs[:, 1]) - abs(pairs[:, 0])
max_vals = [Link](len(pairs))
min_vals = [Link](len(pairs))
pnl = [Link](len(pairs))
base = closes[s]
window = closes[s:e+1]
if start > 0:
max_vals[i] = ([Link]() / base - 1) * 100
min_vals[i] = ([Link]() / base - 1) * 100
pnl[i] = (closes[e] / closes[s] - 1) * 100
else:
max_vals[i] = ([Link]() / base - 1) * -100
min_vals[i] = ([Link]() / base - 1) * -100
pnl[i] = (closes[e] / closes[s] - 1) * -100
patts = crossPatternsLS(df)
patts
Lets gather more patterns and see what our data suggests
first = 0
for symbol in symbols:
sym = symbol + 'USDT'
# print(symbol)
df = kline(sym, tf, 5000)
df['fast'] = ema(df['close'], 50)
df['slow'] = ema(df['close'], 200)
if first == 0:
patts = crossPatternsLS(df)
# print([Link](patts))
first += 1
else:
patts0 = crossPatternsLS(df)
# print([Link](patts0))
patts = [Link]([patts, patts0])
# print([Link](patts))
In [179… t = 1
p = 0.7
In [181… t = 1
p = 0.7
Frankly grouping both long and short strategies into the same patterns might not be worth it, we mostly did it this
way for educational purposes. A cleaner way consists of redefining the function in order to find both and then
keeping them separate:
if mode == 'long':
idxStart = crossover(df['fast'].to_numpy(), df['slow'].to_numpy())
idxEnd = crossover(df['slow'].to_numpy(), df['fast'].to_numpy())
else:
idxStart = crossover(df['slow'].to_numpy(), df['fast'].to_numpy())
idxEnd = crossover(df['fast'].to_numpy(), df['slow'].to_numpy())
closes = df['close'].to_numpy()
lgt = pairs[:, 1] - pairs[:, 0]
max_vals = [Link](len(pairs))
min_vals = [Link](len(pairs))
pnl = [Link](len(pairs))
Multiple Indices
Sometimes a simple condition isn't enough to find a desired pattern, but we have to combine multiple conditions
which might mean having more than one index identifying start or end of our pattern.
Or if we want to run a Machine Learning algorithm to predict the outcome of a pattern two indices won't work, as
we'll need two just to determine the feature range (range where we can save and store features) and an extra index,
effectively defining a second range, that we'll use to get the target value that we're trying to predict for that pattern
(showing the whole pattern to an ML model would be cheating as it also sees the outcome), but we'll tackle this
specific problem further on.
Considering our previous strategy, lets add a step: say we don't want to but when the two moving averages cross but
when the price goes to retest the slower average. We already know how to find the cross indices, we need to figure
out how to find a retest, one idea could be to find all the times price goes below the retest level and then filter out
the indices by selecting the closest index to our crossover:
We notice straight away that this strategy isn't as straight forward as the previous one, where the indices were
inherently sequential and we just had to order them, here we need to filter out loads of idxRetest , we could do
this like this:
nextR = nextR[valid2]
idxS = idxStart[valid2]
s = 300
e = 450
r = 321
plots = [
mpf.make_addplot(df['fast'].iloc[s:e], type= 'line', color= 'grey', width= 3, label= 'fast'),
mpf.make_addplot(df['slow'].iloc[s:e], type= 'line', color= 'black', width= 3, label= 'slow')
]
start_rel = r - s
# Fill the line from start_rel to the end with the level
[Link][start_rel:] = df['low'].iloc[r]
s = 450
e = 800
r = 732
plots = [
mpf.make_addplot(df['fast'].iloc[s:e], type= 'line', color= 'grey', width= 3, label= 'fast'),
mpf.make_addplot(df['slow'].iloc[s:e], type= 'line', color= 'black', width= 3, label= 'slow')
]
start_rel = r - s
# Fill the line from start_rel to the end with the level
[Link][start_rel:] = df['low'].iloc[r]
It appears to be working, but from the second plot we see that considering the retest as low < slow we're leaving
out possibly better entry points, so lets first write a function that finds these patterns and then we can try tweaking
the condition:
In [ ]: def retestPatterns(df):
idxStart = crossover(df['fast'].to_numpy(), df['slow'].to_numpy())
nextR = nextR[valid2]
idxS = idxStart[valid2]
return np.column_stack([idxS, nextR, nextU])
In [ ]: retestPatterns(df)
s = 450
e = 800
r = 637
plots = [
mpf.make_addplot(df['fast'].iloc[s:e], type= 'line', color= 'grey', width= 3, label= 'fast'),
mpf.make_addplot(df['slow'].iloc[s:e], type= 'line', color= 'black', width= 3, label= 'slow')
]
start_rel = r - s
# Fill the line from start_rel to the end with the level
[Link][start_rel:] = df['low'].iloc[r]
So all we did was modify the retest condition to find low prices within a certain range of the slow moving average, i
advise using the mean percent change as it finds where the low price is within one average candle from touching the
slow moving average. We could also apply the same logic but by using (1 + meanChange * m) where m will
represent how many average candles we are from reaching the desired level.
idxS = idxStart[valid]
nextR = nextR[valid]
nextU = nextU[valid]
closes = df['close'].to_numpy()
lgt = nextU - nextR
max_vals = [Link](len(nextR))
min_vals = [Link](len(nextR))
first = 0
for symbol in symbols:
sym = symbol + 'USDT'
print(symbol)
df = kline(sym, tf, 5000)
df['fast'] = ema(df['close'], 50)
df['slow'] = ema(df['close'], 200)
if first == 0:
patts = retestPatterns(df)
print([Link](patts))
first += 1
else:
patts0 = retestPatterns(df)
print([Link](patts0))
patts = [Link]([patts, patts0])
print([Link](patts))
BTC
(12, 7)
ETH
(12, 7)
(24, 7)
SOL
(16, 7)
(40, 7)
In [232… t = 1
p = 0.7
mu_lgt = [Link](patts[:, 3])
mu_min = [Link](patts[:, 4])
mu_max = [Link](patts[:, 5])
mu_pnl = [Link](patts[:, 6])
print(len(patts[patts[:, -1]>0])/len(patts))
First we have to decide how we will define these levels, one idea could be to use the most recent pivot low before the
crossover as a stop loss level and the maximum price reached before our retest as a take profit level.
The maximum price won't require us to find an additional index but the pivot low comes before the crossover in our
logic so we can't look for a minimum between two already found indices:
In [ ]: import [Link]
patts = retestPatternsTPSL(df)
patts
for i in patts:
s = max(0, i[0]-10)
# e = min(len(df)-1, i[3])
e = min(len(df), i[3]+10)
plots = [
mpf.make_addplot(df['fast'].iloc[s:e], type= 'line', color= 'grey', width= 3, label= 'fast'),
mpf.make_addplot(df['slow'].iloc[s:e], type= 'line', color= 'black', width= 3, label= 'slow')
]
hline_data = [
(i[0], df['close'].iloc[i[0]]),
(i[2], df['close'].iloc[i[2]]),
(i[3], df['close'].iloc[i[3]])
]
first = 0
for symbol in symbols:
sym = symbol + 'USDT'
# print(symbol)
df = kline(sym, tf, 5000)
df['fast'] = ema(df['close'], 50)
df['slow'] = ema(df['close'], 200)
if first == 0:
patts = retestPatternsTPSL(df)
# print([Link](patts))
first += 1
else:
patts0 = retestPatternsTPSL(df)
# print([Link](patts0))
patts = [Link]([patts, patts0])
# print([Link](patts))
t = 1
p = 0.7
print(len(patts[patts[:, -1]>0])/len(patts))
55.025 -1.8372973032320936 2.523885499389213 1.2920997444293825
0.35813924798567476 0.7110817061131808 0.5286129033156767
-3.679580688122144 -3.045427412837195 1.0880219945773622 -0.841778196085889
0.65
As we can see we almost doubled the amount of patterns that close at a profit, and the other performance
parameters seem relatively improved overall.
Data Frames
As you add more features or indices, the pattern array becomes harder to read. It might come in handy to save the
patterns as a [Link] to improve readability, unfortunately a lot of the operations you'll run on and with
these patterns require an array type structure (vectors), but you can always convert it back to an array by running
[Link]()
return pattsDf
first = 0
for symbol in symbols:
sym = symbol + 'USDT'
df = kline(sym, tf, 10000)
df['fast'] = ema(df['close'], 50)
df['slow'] = ema(df['close'], 200)
if first == 0:
patts = retestPatternsTPSL(df)
first += 1
else:
patts0 = retestPatternsTPSL(df)
patts = [Link]([patts, patts0])
patts
87 rows × 8 columns