techalyzer/scripts/plotting/plot_signals.py at master · dgunay/techalyzer · GitHub

1
2
3
4
5
6
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
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
# Output objects are usually like this, with some exceptions:
# {
#   "map": {
#     "2020-03-10": {
#       "signal": 0.0,
#       "price": 100.7
#     },
#   },
#  "symbol": "jpm",
#  "indicator": "RelativeStrengthIndex"
# }

import fileinput
import json
import matplotlib.pyplot as plt
import datetime as dt
import pandas as pd
import jsonschema

# TODO: security symbol in title
# TODO: chart security and indicator in subplots
# TODO: buy/sell signals beyond some threshold as vertical lines
# TODO: schema validation of input https://python-jsonschema.readthedocs.io/en/stable/validate/

# schema = {
#      "type" : "object",
#      "properties" : {
#          "price" : {"type" : "number"},
#          "name" : {"type" : "string"},
#      },
#  }

def signals_subplot(ax: plt.Axes, df: pd.DataFrame):
    ax.set_ylim([-1.0, 1.0])
    colors = df['signal'].map(lambda elem: 'red' if elem < 0 else 'green').values
    ax.bar(df.index, df['signal'], color=colors)
    ax.set_title("Bullish/bearish signals")

def indicator_outputs_subplot(ax: plt.Axes, df: pd.DataFrame, indicator: str):
    ta_df = pd.json_normalize(df['output'])
    ta_df.index = df.index
    for col_name, ta_data in ta_df.iteritems():
        ax.plot(ta_data, label=col_name)
    ax.legend(fontsize='x-small', frameon=False)
    ax.set_title(indicator)

def plot_superimposed_indicator(data: dict, df: pd.DataFrame):
    plt.figure()
    fig, (price_ax, sig_ax) = plt.subplots(2, sharex=True)
    fig.suptitle("{} {} signals".format(data['symbol'].upper(), data['indicator']))

    # Price
    price_ax.plot(df['price'])
    price_ax.set_title("{} price".format(data['symbol'].upper()))

    # Technical Indicator outputs
    indicator_outputs_subplot(price_ax, df, data['indicator'])

    # Buy/sell signals
    signals_subplot(sig_ax, df)

    plt.savefig("{}_{}_signal".format(data['symbol'], data['indicator']))

def plot_separate_indicator(data: dict, df: pd.DataFrame):
    plt.figure()
    fig, (price_ax, ta_ax, sig_ax) = plt.subplots(3, sharex=True)
    fig.suptitle("{} {} signals".format(data['symbol'].upper(), data['indicator']))

    # Price
    price_ax.plot(df['price'])
    price_ax.set_title("{} price".format(data['symbol'].upper()))

    # Technical Indicator outputs
    indicator_outputs_subplot(ta_ax, df, data['indicator'])

    # Buy/sell signals
    signals_subplot(sig_ax, df)

    plt.savefig("{}_{}_signal".format(data['symbol'], data['indicator']))


def plot_signals(data: dict):
    # Convert to dataframe with datetime index
    df = pd.DataFrame.from_dict(data['map'], orient='index')
    df.index = pd.to_datetime(df.index)

    if data['indicator'] == 'BollingerBands':
        plot_superimposed_indicator(data, df)
    else:
        plot_separate_indicator(data, df)

####

if __name__ == "__main__":
    json_str = ''
    for line in fileinput.input():
        json_str += line

    try:
        data = json.loads(json_str)
    except Exception as e:
        print("Exception decoding JSON: {}".format(e))

    plot_signals(data)