-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
119 lines (103 loc) · 3.39 KB
/
app.py
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import datetime
import requests
import pandas as pd
from prophet import Prophet
from bs4 import BeautifulSoup as bs
from psx import stocks, tickers
from flask import Flask, jsonify, render_template, request
app = Flask(__name__)
STOCK_NAME = "FFC"
_ = tickers()
data_frame = stocks(
STOCK_NAME, start=datetime.date(2020, 1, 1), end=datetime.date.today()
)
df = pd.DataFrame(
{
"ds": [x.strftime("%Y-%m-%d") for x in data_frame.index],
"y": data_frame["Close"],
}
)
df.index = list(range(len(df)))
forecasting_model = Prophet()
forecasting_model.fit(df)
@app.errorhandler(404)
def page_not_found(_):
"""
Error handler route.
"""
return render_template("error.html"), 404
@app.route("/api/trend")
def trend_api():
"""
API route for stock trends data.
"""
data = {
"x": [
datetime.datetime.utcfromtimestamp(x / 10**9).strftime("%Y-%m-%d")
for x in data_frame.index.to_numpy().tolist()
],
"open": data_frame["Open"].to_numpy().tolist(),
"close": data_frame["Close"].to_numpy().tolist(),
"high": data_frame["High"].to_numpy().tolist(),
"low": data_frame["Low"].to_numpy().tolist(),
"increasing": {"line": {"color": "green"}},
"decreasing": {"line": {"color": "red"}},
"type": "candlestick",
"xaxis": "x",
"yaxis": "y",
}
future = forecasting_model.make_future_dataframe(periods=0)
forecast = forecasting_model.predict(future)
pred = round(forecast.iloc[-1, 1], 2)
return jsonify({"trend": data, "prediction": pred})
@app.route("/")
def home():
"""
Home page route. Returns predictions as well as one-liner live data.
"""
soup = bs(
requests.get(
f"https://dps.psx.com.pk/company/{STOCK_NAME}", timeout=10
).content,
"html.parser",
)
section = soup.find("div", class_="section section--padded company")
lst = [section.find(class_="quote__close").contents[0][3:]]
for stat in section.find_all(class_="stats_item")[:4]:
lst.append(stat.find(class_="stats_value").contents[0])
return render_template(
"home.html",
lst=lst,
time=datetime.date.today().strftime("%Y-%m-%d"),
stock_name=STOCK_NAME,
)
@app.route("/api/date-prediction", methods=["POST"])
def date_prediction_api():
"""
API route for date input prediction.
"""
if request.method == "POST":
if request.form.get("date"):
date = request.form.get("date")
forecast = forecasting_model.predict(pd.DataFrame({"ds": [date]}))
print({"prediction": forecast.iloc[-1, 1]})
return jsonify({"prediction": forecast.iloc[-1, 1]})
return jsonify({"prediction": "Error in getting prediction. Try again."})
return jsonify({"prediction": "Error in getting prediction. Try again."})
@app.route("/date-prediction")
def date_prediction():
"""
Date prediction page route.
"""
return render_template("prediction.html")
@app.route("/api/retrain")
def retrain_model():
"""
API route for triggering model retrain.
"""
global data_frame
data_frame = stocks(
STOCK_NAME, start=datetime.date(2020, 1, 1), end=datetime.date.today()
)
return jsonify({"message": "Training complete. The page will be refreshed."})
app.run(debug=True, use_debugger=False, use_reloader=False, host="0.0.0.0")