Directional changes

What is DC(Directional changes)?

• An approach to summarize price changes other than time series

How to?

• Market prices are sampled based on the DC greater than $P_{EXT}$
• Formally, a DC event is detected when we come across a price $P_c$

Terminology

Extreme point / DCC point

• If the inequality just above holds,

• Note that whilst an extreme point is the end of the one trend, it is also the start point of the next trend
• An extreme point is only recognized in hindsight; precisely at the DCC point

DC event / OS event

• DC means "Directional Change" and OS does "Overshoot"

• Note that for a given time series and a predetermined threshold, the DC summary is unique

Implementation in Python

• Original version

THETA_bakhach=0.1
data_points = market_index['close_scaled'].tolist()
pext = data_points[0]
vals = []
breakpoints = []
yHat = []

for idx, pc in enumerate(data_points):
    if idx == 0:
        continue

    val = (pc - pext)/pext
    dif = pc - pext

	if abs(val) > THETA_bakhach:
        pext = pc
        vals.append(val)
        breakpoints.append(idx)
        yHat.append(data_points[idx])

# endpoint
breakpoints.append(len(data_points) - 1)
vals.append((data_points[-1] - yHat[-1])/yHat[-1])
yHat.append(data_points[-1])

The right-end point is so ugly!! So, add some correction using amount of change .

• Merging adjacent trends

THETA_bakhach=0.1
DELTA_bakhach=(max(market_index['close_scaled'])-min(market_index['close_scaled'])) * 0.08 #0.08
data_points = market_index['close_scaled'].tolist()
pext = data_points[0]
vals = []
breakpoints = []
yHat = []

 
for idx, pc in enumerate(data_points):
    if idx == 0:
        continue

    # val = abs((pc - pext)/pext)
    val = (pc - pext)/pext
    dif = pc - pext

    if abs(val) > THETA_bakhach:
        pext = pc
        vals.append(val)
        breakpoints.append(idx)
        yHat.append(data_points[idx])

# endpoint
breakpoints.append(len(data_points) - 1)
vals.append((data_points[-1] - yHat[-1])/yHat[-1])
yHat.append(data_points[-1])

rm_idx = []
for vi in range(len(vals))[1:]:
    fs = np.sign(vals[vi - 1])
    cs = np.sign(vals[vi])
    if fs == cs:
        rm_idx.append(vi-1)
        
vals_new = [j for i, j in enumerate(vals) if i not in rm_idx]
breakpoints_new = [j for i, j in enumerate(breakpoints) if i not in rm_idx]
yHat_new = [j for i, j in enumerate(yHat) if i not in rm_idx]

The problem in right-end part is solved now! However, it seems that a lot of unnecessary DC has been detected. For this, add a correction merging adjacent trends.

• Add $\delta$ for observing change amount

THETA_bakhach=0.1
DELTA_bakhach=(max(market_index['close_scaled'])-min(market_index['close_scaled'])) * 0.08
data_points = market_index['close_scaled'].tolist()
pext = data_points[0]
vals = []
breakpoints = []
yHat = []

 
for idx, pc in enumerate(data_points):
    if idx == 0:
        continue

    # val = abs((pc - pext)/pext)
    val = (pc - pext)/pext
    dif = pc - pext

    # if abs(dif) > DELTA_bakhach:
    if abs(val) > THETA_bakhach or abs(dif) > DELTA_bakhach:
        pext = pc
        vals.append(val)
        breakpoints.append(idx)
        yHat.append(data_points[idx])

# endpoint
breakpoints.append(len(data_points) - 1)
vals.append((data_points[-1] - yHat[-1])/yHat[-1])
yHat.append(data_points[-1])

rm_idx = []
for vi in range(len(vals))[1:]:
    fs = np.sign(vals[vi - 1])
    cs = np.sign(vals[vi])
    if fs == cs:
        rm_idx.append(vi-1)
        
vals_new = [j for i, j in enumerate(vals) if i not in rm_idx]
breakpoints_new = [j for i, j in enumerate(breakpoints) if i not in rm_idx]
yHat_new = [j for i, j in enumerate(yHat) if i not in rm_idx]

Finally, the detected trends look great!

[References]

• Bakhach, A., Tsang, E.P., & Jalalian, H.R. (2016). Forecasting directional changes in the FX markets. 2016 IEEE Symposium Series on Computational Intelligence (SSCI), 1-8.