Skip to content

utils

candlesticks special

Handy functions for computing various candlesticks patterns from OHLC data

heikinashi

HeikinAshi(japanese_data, ohlc=('timestamp', 'open', 'high', 'low', 'close'))

Computes HeikinAshi Candlesticks Pattern data from Japanese candlesticks pattern data.

Parameters:

Name Type Description Default
japanese_data DataFrame

Pandas DataFrame holding Japanese Candlesticks Pattern Data

required
ohlc tuple

Column names corresponding to 'timestamp', 'open', 'high', 'low' and 'close' data respectively

('timestamp', 'open', 'high', 'low', 'close')

Returns:

Type Description

HeikinAshi Candlesticks Pattern data

Source code in pyalgotrading/utils/candlesticks/heikinashi.py
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def HeikinAshi(japanese_data: pd.DataFrame, ohlc: tuple = ('timestamp', 'open', 'high', 'low', 'close')):
    """
    Computes HeikinAshi Candlesticks Pattern data from Japanese candlesticks pattern data.

    Args:
        japanese_data: Pandas DataFrame holding Japanese Candlesticks Pattern Data
        ohlc: Column names corresponding to 'timestamp', 'open', 'high', 'low' and 'close' data respectively

    Returns:
        HeikinAshi Candlesticks Pattern data

    """
    if not len(ohlc) >= 5:
        print("Argument 'ohlc' should be a tuple of 5 values corresponding to the column names in 'japanese_data' pandas DataFrame, for 'timestamp', 'open', 'high', 'low' and 'close' data respectively.")

    ha_open = 'ha_' + ohlc[1]
    ha_high = 'ha_' + ohlc[2]
    ha_low = 'ha_' + ohlc[3]
    ha_close = 'ha_' + ohlc[4]

    japanese_data = japanese_data.copy()  # make sure we don't modify the original DataFrame
    japanese_data[ha_close] = (japanese_data[ohlc[1]] + japanese_data[ohlc[2]] + japanese_data[ohlc[3]] + japanese_data[ohlc[4]]) / 4

    japanese_data[ha_open] = 0.0
    for i in range(0, len(japanese_data)):
        if i == 0:
            japanese_data[ha_open].iat[i] = (japanese_data[ohlc[1]].iat[i] + japanese_data[ohlc[4]].iat[i]) / 2
        else:
            japanese_data[ha_open].iat[i] = (japanese_data[ha_open].iat[i - 1] + japanese_data[ha_close].iat[i - 1]) / 2

    japanese_data[ha_high] = japanese_data[[ha_open, ha_close, ohlc[2]]].max(axis=1)
    japanese_data[ha_low] = japanese_data[[ha_open, ha_close, ohlc[3]]].min(axis=1)

    # Create separate DataFrame with the required columns only
    heikinashi_data = pd.DataFrame()
    heikinashi_data['timestamp'] = japanese_data['timestamp']
    heikinashi_data['open'] = japanese_data[ha_open]
    heikinashi_data['high'] = japanese_data[ha_high]
    heikinashi_data['low'] = japanese_data[ha_low]
    heikinashi_data['close'] = japanese_data[ha_close]

    return heikinashi_data

linebreak

Linebreak(japanese_candles)

Parameters:

Name Type Description Default
japanese_candles required
Source code in pyalgotrading/utils/candlesticks/linebreak.py
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def Linebreak(japanese_candles):
    """

    Args:
        japanese_candles:

    Returns:

    """
    linebreak_candles = [{'close': candle['close'], 'open': candle['open'], 'timestamp': candle['timestamp']} for _, candle in japanese_candles.iloc[:3].iterrows()]

    for _, candle in japanese_candles.iloc[3:].iterrows():
        all_greater = all(candle['close'] > _ for _ in [max(_linebreakcandle['open'], _linebreakcandle['close']) for _linebreakcandle in linebreak_candles[-3:]])
        all_lesser = all(candle['close'] < _ for _ in [min(_linebreakcandle['open'], _linebreakcandle['close']) for _linebreakcandle in linebreak_candles[-3:]])

        prev_linebreak_candle = linebreak_candles[-1]
        if all_greater:
            new_linebreak_candle = {'open': max(prev_linebreak_candle['open'], prev_linebreak_candle['close']),
                                    'close': candle['close'],
                                    'timestamp': candle['timestamp']}
            linebreak_candles.append(new_linebreak_candle)
        elif all_lesser:
            new_linebreak_candle = {'open': min(prev_linebreak_candle['open'], prev_linebreak_candle['close']),
                                    'close': candle['close'],
                                    'timestamp': candle['timestamp']}
            linebreak_candles.append(new_linebreak_candle)

    return pd.DataFrame(linebreak_candles)

renko

Renko(japanese_candles, brick_count=2, initial_open=None, initial_close=None)

Parameters:

Name Type Description Default
japanese_candles required
brick_count 2
initial_open None
initial_close None
Source code in pyalgotrading/utils/candlesticks/renko.py
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def Renko(japanese_candles, brick_count=2, initial_open=None, initial_close=None):
    """

    Args:
        japanese_candles:
        brick_count:
        initial_open:
        initial_close:

    Returns:

    """
    if not initial_close:
        initial_open = japanese_candles.iloc[0]['open']
        initial_close = japanese_candles.iloc[0]['close']
    else:
        initial_open = japanese_candles.iloc[0]['open'] + (brick_count - ((japanese_candles.iloc[0]['open'] - initial_open) % brick_count))
        initial_close = japanese_candles.iloc[0]['close'] + (brick_count - ((japanese_candles.iloc[0]['close'] - initial_close) % brick_count))

    renko_candles = [{'timestamp': japanese_candles.iloc[0]['timestamp'], 'open': initial_open, 'close': initial_close}]
    prev_renko_candle = renko_candles[-1]

    for _, candle in japanese_candles.iloc[1:].iterrows():
        max_open_close = max(prev_renko_candle['open'], prev_renko_candle['close'])
        min_open_close = min(prev_renko_candle['open'], prev_renko_candle['close'])

        if candle['close'] > max_open_close:
            number_of_renko_candles = math.floor((candle['close'] - max_open_close) / brick_count)
            for i in range(number_of_renko_candles):
                renko_candles.append({'timestamp': candle['timestamp'], 'open': max_open_close, 'close': max_open_close + brick_count})
                prev_renko_candle = renko_candles[-1]
                max_open_close = max(prev_renko_candle['open'], prev_renko_candle['close'])

        elif candle['close'] < min_open_close:
            number_of_renko_candles = math.floor((min_open_close - candle['close']) / brick_count)
            for i in range(number_of_renko_candles):
                renko_candles.append({'timestamp': candle['timestamp'], 'open': min_open_close, 'close': min_open_close - brick_count})
                prev_renko_candle = renko_candles[-1]
                min_open_close = min(prev_renko_candle['open'], prev_renko_candle['close'])

    return pd.DataFrame(renko_candles)

func

A module for plotting candlesticks

import_with_install(package_import_name, package_install_name=None, dependancies=())

Helps import 'package' even if its not installed.

If package is installed, it will be imported and returned. If its not installed, it will be installed using 'pip' and a re-import will be attempted, which should succeed if the package was imported correctly.

Parameters:

Name Type Description Default
package_import_name

name of package to be installed using pip, str

required
package_install_name

name of package to be imported. Default is None, which means package can be imported with the same name as used for installation. If not, this parameter can be used to specify a different import name.

None
dependancies

list of python packages to be installed as additional dependencies'

()

Returns:

Type Description

The imported package

Source code in pyalgotrading/utils/func.py
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def import_with_install(package_import_name, package_install_name=None, dependancies=()):
    """
    Helps import 'package' even if its not installed.

    If package is installed, it will be imported and returned.
    If its not installed, it will be installed using 'pip' and a re-import will be attempted, which should succeed if the package was imported correctly.

    Args:
        package_import_name: name of package to be installed using pip, str
        package_install_name: name of package to be imported. Default is None, which means package can be imported with the same name as used for installation. If not, this parameter can be used to specify a different import name.
        dependancies: list of python packages to be installed as additional dependencies'

    Returns:
        The imported package
    """

    package_install_name = package_install_name if package_install_name is not None else package_import_name

    try:
        return __import__(package_import_name)
    except ImportError:
        print(f"Installing package {package_import_name} via pip. This may take a while...")
        import subprocess
        import sys
        cmd_list = [sys.executable, '-m', 'pip', 'install', package_install_name]
        if dependancies:
            cmd_list.extend(dependancies)
        subprocess.check_call(cmd_list)
        return __import__(package_import_name)

plot_candlestick_chart(data, plot_type, caption='', hide_missing_dates=False, show=True, indicators=(), plot_indicators_separately=False, plot_height=500, plot_width=1000)

Function to create charts for various candlesticks pattern data - - Japanese - Heikin-Ashi - Linebreak - Renko - Japanese for Quandl data

Support for displaying indicator data (on top of candlesticks pattern data or separately).

Parameters:

Name Type Description Default
data DataFrame

Pandas DataFrame with columns timestamp, open, 'high, 'low, close

required
plot_type PlotType

Enum of type PlotType

required
caption str

Caption for the chart

''
hide_missing_dates bool

If True, missing dates in the data (say due to no data over weekend) will be hidden and a continuous plot will be shown. If False, gaps would be shown for missing dates. However, the date formatting on the x-axis is better here, so prefer this when there are no date gaps.

False
show bool

If True, figure will be shown. Useful for displaying figures inline while using Jupyter Notebooks

True
indicators tuple

Indicator data to be displayed

()
plot_indicators_separately bool

If True, indicator data would be plotted in a different subplot. Use this when indicator data range coincides with the historical data range. If False, it will be plotted in the same subplot as the historical data. Use this when indicator data range does not coincide with the historical data range

False
plot_height int

Plot height in pixels

500
plot_width int

Plot width in pixels

1000
Source code in pyalgotrading/utils/func.py
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def plot_candlestick_chart(data: pd.DataFrame, plot_type: PlotType, caption: str = '', hide_missing_dates: bool = False, show: bool = True, indicators: tuple = (), plot_indicators_separately: bool = False, plot_height: int = 500,
                           plot_width: int = 1000):
    """
    Function to create charts for various candlesticks pattern data -
        - Japanese
        - Heikin-Ashi
        - Linebreak
        - Renko
        - Japanese for Quandl data

    Support for displaying indicator data (on top of candlesticks pattern data or separately).

    Args:
        data: Pandas DataFrame with columns `timestamp`, `open`, 'high`, 'low`, `close`
        plot_type: Enum of type PlotType
        caption: Caption for the chart
        hide_missing_dates: If True, missing dates in the `data` (say due to no data over weekend) will be hidden and a continuous plot will be shown.
                            If False, gaps would be shown for missing dates. However, the date formatting on the x-axis is better here, so prefer this when there are no date gaps.
        show: If True, figure will be shown. Useful for displaying figures inline while using Jupyter Notebooks
        indicators: Indicator data to be displayed
        plot_indicators_separately: If True, indicator data would be plotted in a different subplot. Use this when indicator data range coincides with the historical data range.
                                    If False, it will be plotted in the same subplot as the historical data. Use this when indicator data range does not coincide with the historical data range
        plot_height: Plot height in pixels
        plot_width: Plot width in pixels
    """
    import_with_install(package_import_name='plotly', package_install_name='plotly==4.9.0', dependancies=['notebook>=5.3', 'ipywidgets==7.5', 'psutil'])
    # Plotly requirements taken from here: https://pypi.org/project/plotly/
    from plotly.subplots import make_subplots
    from plotly import graph_objects as go

    # Sanity checks
    if not isinstance(plot_type, PlotType):
        print(f'Error: plot_type should be an instance of {PlotType.__class__}')
        return

    # Plot
    if plot_type is PlotType.QUANDL_JAPANESE:
        data['timestamp'] = data.index

    if hide_missing_dates:
        # Plotly has a limitation where if the timestamp are DateTime.DateTime objects which are not continuous,
        # it will plot the missing dates as empty space, which makes the curve look unnatural. The below code gives
        # custom timestamp formatting, which will be the x-axis ticks
        format_timestamp = lambda x: x.strftime("%d/%m %H:%M")
        timestamps = data['timestamp'].apply(format_timestamp)
    else:
        timestamps = data['timestamp']

    candlesticks_data_subplot_row_index = 1
    candlesticks_data_subplot_col_index = 1
    if indicators and (plot_indicators_separately is True):
        fig = make_subplots(rows=3, cols=1, vertical_spacing=0.05, shared_xaxes=True, specs=[[{"rowspan": 2}], [{}], [{}]])
        indicator_subplot_row_index = 3
        indicator_subplot_col_index = 1
    else:
        fig = make_subplots(rows=1, cols=1, vertical_spacing=0.05, shared_xaxes=True)
        indicator_subplot_row_index = 1
        indicator_subplot_col_index = 1

    if plot_type in [PlotType.JAPANESE, PlotType.HEIKINASHI]:
        fig.append_trace(go.Candlestick(x=timestamps, open=data['open'], high=data['high'], low=data['low'], close=data['close'], name='Historical Data'), row=candlesticks_data_subplot_row_index, col=candlesticks_data_subplot_col_index)
    elif plot_type == PlotType.LINEBREAK:
        fig = go.Figure(data=[go.Candlestick(x=timestamps, open=data['open'], high=data[["open", "close"]].max(axis=1), low=data[["open", "close"]].min(axis=1), close=data['close'], name='Historical Data')])
    elif plot_type == PlotType.RENKO:
        fig = go.Figure(data=[go.Candlestick(x=timestamps, open=data['open'], high=data[["open", "close"]].max(axis=1), low=data[["open", "close"]].min(axis=1), close=data['close'], name='Historical Data')])
    elif plot_type == PlotType.QUANDL_JAPANESE:
        fig = go.Figure(data=[go.Candlestick(x=timestamps, open=data['Open'], high=data['High'], low=data['Low'], close=data['Close'], name='Historical Data')])
    else:
        print(f'Error: plot_type ({plot_type}) is not implemented yet')
        return

    for indicator in indicators:
        indicator_name = indicator['name']
        indicator_data = indicator['data']
        extra = indicator['extra'] if 'extra' in indicator else {}
        fig.add_trace(go.Scatter(x=timestamps, y=indicator_data, name=indicator_name, **extra), row=indicator_subplot_row_index, col=indicator_subplot_col_index)

    # Plot customization
    if hide_missing_dates:
        # Plotly has a limitation where if the timestamp are DateTime.DateTime objects which are not continuous,
        # it will plot the missing dates as empty space, which makes the curve look unnatural. Hence, the below fix
        fig.layout.xaxis.type = 'category'

    fig.update(layout_xaxis_rangeslider_visible=False)
    fig.update_layout(title={'text': caption, 'y': 0.9, 'x': 0.5, 'xanchor': 'center', 'yanchor': 'bottom'}, height=plot_height, width=plot_width)

    # Show the plot
    if show:
        fig.show()

Last update:

Comments