In [156]:
from IPython.display import HTML
HTML('''<script> code_show=true;  
function code_toggle() {
 if (code_show){  $('div.input').hide(); 
 } else {  $('div.input').show();  }
 code_show = !code_show
}
$( document ).ready(code_toggle); </script>
The raw code for this IPython notebook is by default hidden for easier reading.
To toggle on/off the raw code, click <a href="javascript:code_toggle()">here</a>.''')


Out[156]:
The raw code for this IPython notebook is by default hidden for easier reading. To toggle on/off the raw code, click here.

In [10]:
cd ..


/Users/miguel/Jottacloud/devel/osqf2015

In [6]:
from datetime import datetime, date

In [7]:
from IPython.html.widgets import interact, interactive, fixed


:0: FutureWarning: IPython widgets are experimental and may change in the future.

In [8]:
%reload_ext autoreload
%autoreload 2

In [92]:
from model import *

In [12]:
from cytoolz import sliding_window, count, pipe, thread_first

In [13]:
import seaborn as sns
%matplotlib notebook

In [23]:
from bokeh.plotting import *
from bokeh.models import TapTool

output_notebook()


BokehJS successfully loaded.

Load risk factors using Blaze

They correspond to

  • Deutsche Bank (in USD)
  • EURUSD

In [18]:
rf1 = RiskFactor.from_blaze("data/db2.bcolz")
rfeurusd = CurrencyRiskFactor.from_blaze("data/eurusd.bcolz")

Perform the Filtered Historical Simulation


In [22]:
rf1.logreturns()
rf1.devol()
scenarios = rf1.fhs()
df = pd.DataFrame(scenarios, index=rf1.ts.index)

In [20]:
metric = VaR(99)

Computation functions


In [180]:
def setup_results(df):
    results = pd.DataFrame({'neutral': df.icol(0),
                            'realized': df.shift(-1).icol(0)},
                            index=df.index )
    return results

def quantile_results(results, df, level):
    n_scenarios = len(df.columns) - 1
    results['q_upper'] = df.icol(np.arange(1,n_scenarios+1)).apply(np.percentile, axis=1, q=level)
    results['q_lower'] = df.icol(np.arange(1,n_scenarios+1)).apply(np.percentile, axis=1, q=100.-level)
    results.eval('shortfall = q_lower - realized')
    results.eval('shortfall_mid = shortfall * 0.5')
    return results

def mark_outliers(df, level):
    is_outlier = df.eval('(q_lower > realized)')
    df.loc[is_outlier,'bad'] = df[is_outlier]['realized']
    is_outlier = df.eval('q_upper < realized')
    df.loc[is_outlier,'good'] = df[is_outlier]['realized']
    
    n_outliers = pd.stats.moments.rolling_count(df.bad, 250)
    df['outliers'] = n_outliers
    m = VaR(level/100.)
    stat = m.likelihood_statistic(n_outliers, 250)
    df['lights'] = VaR.traffic_light(m.confidence(stat))
    return df

In [181]:
def process_full(df, level=99, start='2010-01-01'):
    df_restricted = df[df.index > start]
    return thread_first(df_restricted,
                        setup_results,
                        (quantile_results, df_restricted, level),
                        (mark_outliers, level),
                        lambda x: x.ix[:-2])

Generation of a ColumnDataSource for a plot


In [182]:
def append_reverse(a, b):
    return np.append(a, np.flipud(b))

def make_envelope_ds(df, coordinates='relative'):
    """ construct data source """
    src = ColumnDataSource()
    d = df.index.tolist()
    src.add(append_reverse(d,d), 'date')
    return src

def update_envelope_ds(df, src, coordinates='relative'):
    """ update data source """
    factor = 0. if coordinates == 'absolute' else df.neutral
    y = append_reverse((df.q_lower - factor).values,
                       (df.q_upper - factor).values)
    src.add(y.tolist(), 'var')
    src.add((df.good - factor).values, 'good')
    src.add((df.bad - factor).values, 'bad')
    src.add((df.realized - factor).values, 'realized')
    return src

def make_outlier_ds(df):
    src = ColumnDataSource()
    src.add( df.index.tolist(), 'date')
    return src

def update_outlier_ds(df, src):
    src.add( df.outliers, 'outliers')
    src.add( df.outliers * 0.5, 'outliers_mids')
    colors = np.array(['green', 'yellow', 'red'])
    src.add(colors[df.lights], 'lights')
    return src

Plot creation


In [183]:
def make_envelope_plot(src):
    p = figure(title="Backtesting outliers", plot_height=400, plot_width=900, x_axis_type="datetime")
    p.patch('date', 'var', source=src, color='grey', fill_alpha=0.2)
    p.line('date', 'realized', source=src, line_width=1, color='steelblue')
    p.scatter(x='date', y='good', source=src, size=10, color='green', fill_alpha=0.2, )
    p.scatter(x='date', y='bad', source=src, size=10, color='red', fill_alpha=0.2, )
    return p

In [184]:
def make_outlier_plot(src, range_obj):
    p = figure(plot_height=200, plot_width=900, x_axis_type="datetime", x_range=range_obj)
    w = 20*60*60*1000 # half day in ms
    p.rect(x="date", y='outliers_mids', width=w, height='outliers', color='lights', fill_alpha=0.3, source=src)
    return p

In [185]:
df_res = process_full(df)

In [187]:
ds_envelope = make_envelope_ds(df_res)
ds_envelope = update_envelope_ds(df_res, ds_envelope)

In [190]:
ds_outliers = make_outlier_ds(df_res)
ds_outliers = update_outlier_ds(df_res, ds_outliers)

In [191]:
p = make_envelope_plot(ds_envelope)
p2 = make_outlier_plot(ds_outliers, p.x_range)
show(gridplot( [[p],[p2]]))



In [199]:
interact(update_confidence_level, confidence_level=(90, 99, 1));
interact(update_lambda, _lambda=(0.80, 0.99, 0.01));



In [197]:
def update_confidence_level(confidence_level=94, cfg=fixed(c)):
    cfg['confidence_level'] = confidence_level
    df_res = process_full(df, level=cfg['confidence_level'])
#     print( df_res.tail())
    ds = update_envelope_ds(df_res, cfg['src'])
    ds_outliers = update_outlier_ds(df_res, cfg['src_out'])
    # propagate changes to plot
    ds.push_notebook()
    ds_outliers.push_notebook()

In [198]:
def update_lambda(_lambda=0.94, cfg=fixed(c)):
    cfg['l'] = 1. - _lambda
    rf1.devol(_lambda=1. - _lambda)
    df = pd.DataFrame(rf1.fhs(), index=rf1.ts.index)
    cfg['df'] = df
    df_res = process_full(df, level=cfg['confidence_level'])
    ds = update_envelope_ds(df_res, cfg['src'])
    ds_outliers = update_outlier_ds(df_res, cfg['src_out'])
    # propagate changes to plot
    ds.push_notebook()
    ds_outliers.push_notebook()

In [196]:
# import collections
# Config = collections.namedtuple('Config', 'l confidence_level src df')
c = dict(confidence_level=99, l=0.94, src=ds_envelope, src_out=ds_outliers, df=df)

In [98]:


In [ ]:


In [ ]:


In [ ]:


In [ ]:


In [ ]:


In [ ]:


In [ ]:


In [ ]:


In [ ]:

Testing Pandas Panel


In [119]:
pos = pd.Panel({'eurusd': rfeurusd.ts, 'db': rf1.ts})

In [120]:
pos['db',1:10,:]


Out[120]:
Value LogReturns Vola DevolLogReturns
Date
1996-11-19 47.87 0.007759 NaN NaN
1996-11-20 47.50 -0.007759 0.010973 -0.707107
1996-11-21 48.00 0.010471 0.009887 1.059066
1996-11-22 47.37 -0.013212 0.011737 -1.125696
1996-11-25 47.50 0.002741 0.010157 0.269817
1996-11-26 47.75 0.005249 0.009244 0.567887
1996-11-27 46.75 -0.021165 0.012218 -1.732230
1996-11-29 47.62 0.018439 0.013843 1.331959
1996-12-02 47.00 -0.013105 0.013674 -0.958375

In [162]:
indexer = (pos.major_axis > to_dt( '2010-01-01')) & (pos.major_axis < to_dt( '2015-01-01'))
pp = pos.take(indexer, axis='major')
# pos.take( indexer, axis='major')

In [178]:
pos.iloc[:, indexer,:]


Out[178]:
<class 'pandas.core.panel.Panel'>
Dimensions: 2 (items) x 1303 (major_axis) x 4 (minor_axis)
Items axis: db to eurusd
Major_axis axis: 2010-01-04 00:00:00 to 2014-12-31 00:00:00
Minor_axis axis: Value to DevolLogReturns