aboutsummaryrefslogtreecommitdiffstats
path: root/python/intex/intex_scenarios.py
diff options
context:
space:
mode:
Diffstat (limited to 'python/intex/intex_scenarios.py')
-rw-r--r--python/intex/intex_scenarios.py148
1 files changed, 99 insertions, 49 deletions
diff --git a/python/intex/intex_scenarios.py b/python/intex/intex_scenarios.py
index 75da23a6..79b50639 100644
--- a/python/intex/intex_scenarios.py
+++ b/python/intex/intex_scenarios.py
@@ -27,8 +27,9 @@ pattern9 = re.compile("(?P<a>SEVERITY\[\w+,\d+\]=)mkt\(70\)")
global_reinvfloatpercentage = 84
global_reinvfixedpercentage = 16
+
def get_reinv_assets(conn, dealname, workdate):
- sqlstr = 'SELECT * FROM et_historicaldealinfo(%s, %s) WHERE ReinvFlag IS TRUE'
+ sqlstr = "SELECT * FROM et_historicaldealinfo(%s, %s) WHERE ReinvFlag IS TRUE"
d = {}
with conn.cursor() as c:
c.execute(sqlstr, (dealname, workdate))
@@ -37,10 +38,13 @@ def get_reinv_assets(conn, dealname, workdate):
conn.commit()
return d
+
def get_recovery(conn, dealname, workdate, defaultrecovery=50):
""" compute average price of defaulted assets """
- sqlstr = ("select sum(coalesce(price, %s) * currentbalance)/sum(currentbalance) AS recov "
- "from et_aggdealinfo_historical(%s, %s) where defaultedflag is True")
+ sqlstr = (
+ "select sum(coalesce(price, %s) * currentbalance)/sum(currentbalance) AS recov "
+ "from et_aggdealinfo_historical(%s, %s) where defaultedflag is True"
+ )
with conn.cursor() as c:
try:
c.execute(sqlstr, (defaultrecovery, dealname, workdate))
@@ -53,8 +57,9 @@ def get_recovery(conn, dealname, workdate, defaultrecovery=50):
conn.commit()
return float(recovery)
+
def get_reinvenddate(conn, dealname, workdate):
- sqlstr = 'SELECT reinv_end_date FROM historical_clo_universe(%s, %s)'
+ sqlstr = "SELECT reinv_end_date FROM historical_clo_universe(%s, %s)"
with conn.cursor() as c:
c.execute(sqlstr, (dealname, workdate))
reinvenddate, = c.fetchone()
@@ -64,10 +69,13 @@ def get_reinvenddate(conn, dealname, workdate):
else:
raise Exception("missing reinvestment end date")
+
def generate_scenarios(workdate, dealname, conn):
- prometheus = os.path.join(os.environ['BASE_DIR'], "Scenarios", "prometheus.sss")
+ prometheus = os.path.join(os.environ["BASE_DIR"], "Scenarios", "prometheus.sss")
n_scenarios = 100
- basedir = os.path.join(os.environ['BASE_DIR'], "Scenarios", "Intex curves_" + workdate)
+ basedir = os.path.join(
+ os.environ["BASE_DIR"], "Scenarios", "Intex curves_" + workdate
+ )
defaultedprice = get_recovery(conn, dealname, workdate)
replace = "\g<a>{0:.3f}".format(defaultedprice)
try:
@@ -76,22 +84,26 @@ def generate_scenarios(workdate, dealname, conn):
except IOError:
logger.error("{0}: config file doesn't exist".format(dealname))
return
- reinvflag = config['reinvflag']
+ reinvflag = config["reinvflag"]
if reinvflag:
reinvenddate = get_reinvenddate(conn, dealname, workdate)
reinv_assets = get_reinv_assets(conn, dealname, workdate)
- n_float_assets = len([v for v in reinv_assets.values() if v[0] == 'FLOAT'])
- n_fixed_assets = len([v for v in reinv_assets.values() if v[0] == 'FIXED'])
- rollingmat = config['rollingmat']
+ n_float_assets = len([v for v in reinv_assets.values() if v[0] == "FLOAT"])
+ n_fixed_assets = len([v for v in reinv_assets.values() if v[0] == "FIXED"])
+ rollingmat = config["rollingmat"]
if n_fixed_assets == 0:
reinvfixedpercentage = 0
else:
reinvfixedpercentage = global_reinvfixedpercentage / n_fixed_assets
if n_float_assets > 0:
- reinvfloatpercentage = (100 - n_fixed_assets * reinvfixedpercentage)/n_float_assets
+ reinvfloatpercentage = (
+ 100 - n_fixed_assets * reinvfixedpercentage
+ ) / n_float_assets
try:
- with open(os.path.join(basedir, "csv", dealname + "-reinvprices.csv"), "r") as fh:
+ with open(
+ os.path.join(basedir, "csv", dealname + "-reinvprices.csv"), "r"
+ ) as fh:
dr = csv.DictReader(fh)
reinvprices = {f: [] for f in dr.fieldnames}
for line in dr:
@@ -99,7 +111,9 @@ def generate_scenarios(workdate, dealname, conn):
try:
val = float(line[f])
except ValueError:
- logger.error("Incorrect value in reinvprices for {}".format(dealname))
+ logger.error(
+ "Incorrect value in reinvprices for {}".format(dealname)
+ )
else:
reinvprices[f].append("{0:.3f}".format(val))
@@ -119,15 +133,17 @@ def generate_scenarios(workdate, dealname, conn):
cdrline = next(csvcdr)
cdrline = "\t".join(["{0:.3f}".format(float(cdr)) for cdr in cdrline])
recoveryline = next(csvrecovery)
- recoveryline = "\t".join(["{0:.3f}".format(float(recovery)) for recovery in recoveryline])
+ recoveryline = "\t".join(
+ ["{0:.3f}".format(float(recovery)) for recovery in recoveryline]
+ )
- i=1
+ i = 1
with open(prometheus) as fh:
for line in fh:
line = line.rstrip()
if "DEAL_NAME" in line:
- newline = "DEAL_NAME=" + dealname.upper() + "\r\n"
+ newline = "DEAL_NAME=" + dealname.upper() + "\r\n"
fhsss.write(newline)
continue
if not reinvflag and pattern8.match(line):
@@ -147,33 +163,39 @@ def generate_scenarios(workdate, dealname, conn):
coupon = 4
elif fixedorfloat == "FIXED":
coupon = 7
- line = "REINVEST[{0}::REINV_TBA{1}][DEAL,{2}]=".format(dealname.upper(),
- reinv_number,
- scen_number)
- line += "COUP_SPR={0}|AMORT=Bullet|USE_REINVEST_PIP=1|MAT_DATE={1}|".format(coupon, rollingmat)
+ line = "REINVEST[{0}::REINV_TBA{1}][DEAL,{2}]=".format(
+ dealname.upper(), reinv_number, scen_number
+ )
+ line += "COUP_SPR={0}|AMORT=Bullet|USE_REINVEST_PIP=1|MAT_DATE={1}|".format(
+ coupon, rollingmat
+ )
if liborfloor is not None:
line += "|ARM_LIFE_FLOOR={}".format(liborfloor)
fhsss.write(line + "\r\n")
continue
if reinvflag and pattern2.match(line):
- line = re.sub(pattern2, r"\1{0}\2{1}", line).format(dealname.upper(), reinvenddate)
+ line = re.sub(pattern2, r"\1{0}\2{1}", line).format(
+ dealname.upper(), reinvenddate
+ )
fhsss.write(line + "\r\n")
continue
m = pattern3.match(line)
if reinvflag and m:
reinv_number, scen_number = m.groups()
- if dealname=="litpt3" and reinv_number=="1":
- line = "STANDARD_VAR[LITPT3::#REINVLOANP100,{0}]".format(scen_number) + \
- "={0}".format(" ".join(reinvprices["REINV_TBA1"]))
+ if dealname == "litpt3" and reinv_number == "1":
+ line = "STANDARD_VAR[LITPT3::#REINVLOANP100,{0}]".format(
+ scen_number
+ ) + "={0}".format(" ".join(reinvprices["REINV_TBA1"]))
fhsss.write(line + "\r\n")
continue
reinv_name = "REINV_TBA" + reinv_number
if reinv_name in reinvprices:
- line = \
- "STANDARD_VAR[{0}::#PRICE100_TBA{1},{2}]={3}".format(dealname.upper(),
- reinv_number,
- scen_number,
- " ".join(reinvprices[reinv_name]))
+ line = "STANDARD_VAR[{0}::#PRICE100_TBA{1},{2}]={3}".format(
+ dealname.upper(),
+ reinv_number,
+ scen_number,
+ " ".join(reinvprices[reinv_name]),
+ )
fhsss.write(line + "\r\n")
continue
m = pattern5.match(line)
@@ -182,24 +204,32 @@ def generate_scenarios(workdate, dealname, conn):
reinv_name = "REINV_TBA" + reinv_number
if reinv_number == "1":
if dealname == "litpt3":
- line = "STANDARD_VAR[LITPT3::#LOANREINVPCT,{0}]=100".format(scen_number)
+ line = "STANDARD_VAR[LITPT3::#LOANREINVPCT,{0}]=100".format(
+ scen_number
+ )
fhsss.write(line + "\r\n")
continue
if dealname == "flags4":
- line = "STANDARD_VAR[FLAGS4::#PCT100_TBA1,{0}]=100".format(scen_number)
+ line = "STANDARD_VAR[FLAGS4::#PCT100_TBA1,{0}]=100".format(
+ scen_number
+ )
fhsss.write(line + "\r\n")
continue
if reinv_name in reinv_assets:
- if reinv_assets[reinv_name][0] == 'FIXED':
- line = "STANDARD_VAR[{0}::#REINVPCT_TBA{1},{2}]={3}".format(dealname.upper(),
- reinv_number,
- scen_number,
- reinvfixedpercentage)
- elif reinv_assets[reinv_name][0] == 'FLOAT':
- line = "STANDARD_VAR[{0}::#REINVPCT_TBA{1},{2}]={3}".format(dealname.upper(),
- reinv_number,
- scen_number,
- reinvfloatpercentage)
+ if reinv_assets[reinv_name][0] == "FIXED":
+ line = "STANDARD_VAR[{0}::#REINVPCT_TBA{1},{2}]={3}".format(
+ dealname.upper(),
+ reinv_number,
+ scen_number,
+ reinvfixedpercentage,
+ )
+ elif reinv_assets[reinv_name][0] == "FLOAT":
+ line = "STANDARD_VAR[{0}::#REINVPCT_TBA{1},{2}]={3}".format(
+ dealname.upper(),
+ reinv_number,
+ scen_number,
+ reinvfloatpercentage,
+ )
fhsss.write(line + "\r\n")
continue
@@ -217,10 +247,20 @@ def generate_scenarios(workdate, dealname, conn):
i = i + 1
if i <= n_scenarios:
cdrline = next(csvcdr)
- cdrline = "\t".join(["{0:.3f}".format(float(cdr)) for cdr in cdrline]) + "\r\n"
+ cdrline = (
+ "\t".join(["{0:.3f}".format(float(cdr)) for cdr in cdrline])
+ + "\r\n"
+ )
recoveryline = next(csvrecovery)
- recoveryline = "\t".join(["{0:.3f}".format(float(recovery)) \
- for recovery in recoveryline]) + "\r\n"
+ recoveryline = (
+ "\t".join(
+ [
+ "{0:.3f}".format(float(recovery))
+ for recovery in recoveryline
+ ]
+ )
+ + "\r\n"
+ )
continue
if "LOSS_NONPERF_SEVERITY" in line:
line = re.sub(pattern9, replace, line)
@@ -230,9 +270,11 @@ def generate_scenarios(workdate, dealname, conn):
fhcdr.close()
logger.info("generated scenarios for: {0}".format(dealname))
+
if __name__ == "__main__":
sys.path.append(".")
from utils.db import dbconn
+
if len(sys.argv) > 1:
workdate = sys.argv[1]
else:
@@ -240,11 +282,19 @@ if __name__ == "__main__":
if len(sys.argv) > 2:
dealnames = sys.argv[2:]
else:
- dealnames = [d.split(".")[0] for d in
- os.listdir(os.path.join(os.environ['BASE_DIR'], "Scenarios",
- "Intex curves_" + workdate, "csv"))
- if "RData" in d]
- ET = dbconn('etdb')
+ dealnames = [
+ d.split(".")[0]
+ for d in os.listdir(
+ os.path.join(
+ os.environ["BASE_DIR"],
+ "Scenarios",
+ "Intex curves_" + workdate,
+ "csv",
+ )
+ )
+ if "RData" in d
+ ]
+ ET = dbconn("etdb")
for dealname in dealnames:
generate_scenarios(workdate, dealname, ET)
ET.close()
generated by cgit v1.2.3-70-g09d2 (git 2.52.0) at 2025-12-22 04:28:20 +0000