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
|
import csv
import datetime
import os
import psycopg2
import yaml
from risk_insight import app
from flask import render_template, jsonify, request, g
from psycopg2 import sql
def get_attach_from_name(index_type, series):
if index_type.lower() == "ig":
if series == 9:
attach = [0, 3, 7, 10, 15, 30, 100]
else:
attach = [0, 3, 7, 15, 100]
elif index_type.lower() == "hy":
attach = [0, 15, 25, 35, 100]
elif index_type.lower() == "xo":
attach = [0, 10, 20, 35, 100]
elif index_type.lower() == "eu":
if series == 9:
attach = [0, 3, 6, 9, 12, 22, 100]
else:
attach = [0, 3, 6, 12, 100]
return attach
def get_db():
db = getattr(g, '_database', None)
if db is None:
db = g._database = psycopg2.connect(database="serenitasdb",
user="serenitas_user",
host="debian")
return db
@app.teardown_appcontext
def close_connection(exception):
db = getattr(g, '_database', None)
if db is not None:
db.close()
@app.route("/_data_tranches")
def get_risk_numbers():
index = request.args.get("i")
series = request.args.get("s", 0, int)
tenor = request.args.get("t")
greek = request.args.get("g")
db = get_db()
attach = get_attach_from_name(index, series)
sqlstr = (sql.SQL("SELECT to_char(date, 'YYYY/MM/DD'), array_agg({}) FROM " \
"(SELECT date, {} FROM risk_numbers_new " \
"WHERE index=%s AND series=%s AND tenor=%s ORDER BY date, attach) "\
"AS a GROUP BY date ORDER BY date").
format(sql.Identifier(greek), sql.Identifier(greek)))
with db.cursor() as c:
c.execute(sqlstr, (index.upper(), series, tenor))
data = [[date] + arr for date, arr in c]
return jsonify(labels=["Date"] + ["{0}-{1} {2}".format(l, u, greek) for l, u in zip(attach[:-1], attach[1:])],
data=data)
@app.route("/_data_indices")
def get_indices_quotes():
index = request.args.get("i")
series = request.args.get("s", 0, int)
tenor = request.args.get("t")
what = request.args.get("w")
data = []
db = get_db()
sqlstr = "SELECT to_char(date, 'YYYY/MM/DD'), {0} " \
"from index_quotes WHERE index=%s and series=%s " \
"and tenor=%s ORDER BY date"
query_columns = {'quotes': "closeprice, modelprice",
'adjquotes': "adjcloseprice, adjmodelprice",
'basis': "closeprice-modelprice",
'theta': "theta",
'duration': "duration"}
labels = {'quotes': ["Date", "Close", "FV"],
'adjquotes': ["Date", "AdjClose", "AdjFV"],
'basis': ["Date", "Basis"],
'theta': ["Date", "Theta"],
'duration': ["Date", "Duration"]}
sqlstr = sqlstr.format(query_columns[what])
with db.cursor() as c:
c.execute(sqlstr, (index.upper(), series, tenor))
data = c.fetchall()
return jsonify(labels = labels[what],
data = data)
@app.route("/tranches.html")
def tranches():
with open(os.path.join(os.getenv("CODE_DIR"), "etc", "runs.yml")) as fh:
config = yaml.load(fh)
series_list = sorted(list(set([int(index[2:]) for index, tenor in config['runs']])))
return render_template("tranches.html", series=series_list)
@app.route("/indices.html")
def indices():
return render_template("indices.html", series=[9, 10, 11, 13] + list(range(16, 30)))
@app.route("/")
def main():
return render_template("index.html")
|
