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
120
121
122
123
124
125
126
127
128
129
130
131
|
import pandas as pd
import requests
import time
from bs4 import BeautifulSoup
from utils.db import dbconn
from streamz import Stream
def combine_df(state, new):
state = state.append(df)
state = state.drop_duplicates(["Dissemination Id"])
return state
def event_loop(conn):
url = "https://kgc0418-tdw-data-0.s3.amazonaws.com/prices/CREDITS_PRICE_DETAIL.HTML"
df_indic = get_index_indicative(conn)
acc = pd.DataFrame()
i = 0
old_ts = None
while i < 10000:
i += 1
r = requests.get(url)
if r.status_code == 200:
soup = BeautifulSoup(r.content, features="lxml")
df = parse_soup(soup)
acc = acc.append(df)
acc = acc.drop_duplicates(["Dissemination Id"])
# df_options = df[df["UPI/Taxonomy"].str.startswith("Credit:Swaptions")]
df_index = acc[acc["UPI/Taxonomy"].str.startswith("Credit:Index")]
df_index = parse_index(df_index, df_indic)
last_trade = df_index.xs(("IG", 32, 1, "5yr"))
last_ts = last_trade.index[-1]
if old_ts is None or last_ts > old_ts:
print(last_ts, last_trade["Price Notation"].iloc[-1])
old_ts = last_ts
time.sleep(0.5)
return acc
def get_index_indicative(conn):
return pd.read_sql_query(
"SELECT redindexcode, index, series, version, tenor, maturity FROM index_desc",
conn,
parse_dates=["maturity"],
)
def parse_soup(soup):
table = soup.find_all("table")[1]
rows = iter(table.find_all("tr"))
header = [th.text for th in next(rows).find_all("th")]
rows = [[td.text.strip() or None for td in r.find_all("td")] for r in rows]
df = pd.DataFrame(rows, columns=header)
df["Dissemination Id"] = df["Dissemination Id"].astype("int")
df["Original Dissemination Id"] = (
df["Original Dissemination Id"].astype("float").astype("Int64")
)
for col in ["Execution Timestamp", "End Date"]:
df[col] = pd.to_datetime(df[col])
df["Execution Timestamp"] = (
df["Execution Timestamp"]
.dt.tz_localize("utc")
.dt.tz_convert("America/New_York")
)
df["Underlying Asset 1"] = df["Underlying Asset 1"].str.rsplit(
":", n=1, expand=True
)[1]
return df
def parse_index(df, df_indic):
df = df.rename(
columns={
"Execution Timestamp": "trade_timestamp",
"Underlying Asset 1": "redindexcode",
"End Date": "maturity",
}
)
df = df.merge(df_indic, on=["redindexcode", "maturity"])
df = df.set_index(
["index", "series", "version", "tenor", "trade_timestamp"]
).sort_index()
df["Price Notation"] = pd.to_numeric(df["Price Notation"])
return df
def parse_options(df, df_indic):
for col in [
"Action",
"Cleared or Uncleared",
"Price Notation Type",
"Option Type",
"Option Currency",
"Day Count Convention",
]:
df[col] = df[col].astype("category")
for col in ["Option Premium"]:
df[col] = pd.to_numeric(df[col].str.replace(",", ""))
for col in ["Option Strike Price", "Price Notation"]:
df[col] = df[col].astype("float")
df = df.rename(
columns={
"Option Strike Price": "strike",
"Option Expiration Date": "expiration_date",
"Underlying Asset 1": "redindexcode",
"Rounded Notional Amount1": "notional",
"Option Premium": "premium",
"Option Type": "option_type",
"Price Notation": "price",
"Execution Timestamp": "trade_timestamp",
}
)
df.strike = df.strike.where(df.strike < 1000, df.strike / 100).where(
df.strike > 10, df.strike * 100
)
df.price = (df.price * 1e2).where(
df["Price Notation Type"] == "Percentage", df.price
)
df = df.merge(df_indic, on="redindexcode")
df = df.set_index(["index", "series", "version", "trade_timestamp"]).sort_index()
return df[
["expiration_date", "notional", "strike", "option_type", "premium", "price"]
]
if __name__ == "__main__":
conn = dbconn("serenitasdb")
df = event_loop(conn)
conn.close()
|
