var-demo.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Value-at-risk calculations\n",
    "\n",
    "_Do not use this code to guide actual investment decisions!_\n",
    "\n",
    "## Basic setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql import SparkSession, SQLContext\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Loading the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "df = spark.read.load(\"/data/wikieod8.parquet\").cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Calculating historical returns\n",
    "\n",
    "We'll use Spark's windowing functions over data frames to determine the percentage change in each security's price from the previous close to each day's close."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import Window\n",
    "from pyspark.sql.functions import lag, col, avg, variance\n",
    "\n",
    "ddf = df.select(\"ticker\", \"date\", \"close\").withColumn(\"change\", (col(\"close\") / lag(\"close\", 1).over(Window.partitionBy(\"ticker\").orderBy(df[\"date\"])) - 1.0) * 100)\n",
    "ddf.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Characterizing expected return distributions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import sqrt\n",
    "mv = ddf.groupBy(\"ticker\").agg(avg(\"change\").alias(\"mean\"), sqrt(variance(\"change\")).alias(\"stddev\"))\n",
    "mv.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since there are only about 3,000 ticker symbols in our data set, we can easily collect these in driver memory for use in our simulation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dist_map = mv.rdd.map(lambda r: (r[0], (r[1], r[2]))).collectAsMap()\n",
    "dist_map[\"RHT\"]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Getting current security prices"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import first\n",
    "priceDF = ddf.orderBy(\"date\", ascending=False).groupBy(\"ticker\").agg(first(\"close\").alias(\"price\"), first(\"date\").alias(\"date\")).cache()\n",
    "priceDF.show(10)\n",
    "\n",
    "prices = priceDF.rdd.map(lambda r: (r[0], r[1])).collectAsMap()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setting up a simulation\n",
    "\n",
    "### Generating a random portfolio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from random import randint, seed\n",
    "\n",
    "def random_portfolio(symbols):\n",
    "    result = {}\n",
    "    for s in symbols:\n",
    "        result[s] = prices[s] * (randint(1, 1000) * 11)\n",
    "    return result\n",
    "\n",
    "def portfolio_value(pf):\n",
    "    return sum([v for v in pf.values()])\n",
    "\n",
    "seed(0xdea110c8)\n",
    "\n",
    "portfolio = random_portfolio(ddf.select(\"ticker\").distinct().sample(True, 0.01, 0xdea110c8).rdd.map(lambda r: r[0]).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generating random seeds for each simulation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def seeds(count):\n",
    "    return [randint(0, 1 << 32 - 1) for i in range(count)]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Defining the simulation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def simstep(pf, params, prng):\n",
    "    def daily_return(sym):\n",
    "        mean, stddev = params[sym]\n",
    "        change = (prng.normalvariate(mean, stddev) + 100) / 100.0\n",
    "        return change\n",
    "    return dict([(s, daily_return(s) * v) for s, v in pf.items()])\n",
    "\n",
    "def simulate(seed, pf, params, days):\n",
    "    from random import Random\n",
    "    prng = Random()\n",
    "    prng.seed(seed)\n",
    "    pf = pf.copy()\n",
    "    \n",
    "    for day in range(days):\n",
    "        pf = simstep(pf, params, prng)\n",
    "    return pf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Simulating five days of market activity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "days_to_simulate = 5\n",
    "simulation_count = 10000\n",
    "\n",
    "sc = spark.sparkContext\n",
    "seed_rdd = sc.parallelize(seeds(simulation_count))\n",
    "bparams = sc.broadcast(dist_map)\n",
    "bpf = sc.broadcast(portfolio)\n",
    "initial_value = portfolio_value(portfolio)\n",
    "\n",
    "results = seed_rdd.map(lambda s: portfolio_value(simulate(s, bpf.value, bparams.value, days_to_simulate)) - initial_value)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "simulated_results = list(zip(results.collect(), seed_rdd.collect()))\n",
    "simulated_values = [v for (v, _) in simulated_results]\n",
    "simulated_values.sort()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "%config InlineBackend.figure_format = 'svg'\n",
    "\n",
    "import seaborn as sns\n",
    "import numpy as np\n",
    "sns.set(color_codes=True)\n",
    "\n",
    "_ = sns.distplot(simulated_values, kde=False).set(xlabel=\"$ change\", ylabel=\"simulation count\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# plot only every 20th simulation result (unless that would leave us with fewer \n",
    "# than 50 elements) to avoid generating a huge figure\n",
    "plotevery = len(simulated_values) > 1000 and 20 or 1\n",
    "\n",
    "xvals = [float(i) / len(simulated_values) for i in range(len(simulated_values))]\n",
    "ax = sns.tsplot(np.array(simulated_values[::plotevery]), np.array(xvals[::plotevery]))\n",
    "\n",
    "# add ticks for every ten percent\n",
    "ax.get_xaxis().set_ticks([i * 0.1 for i in range(11)])\n",
    "_ = ax.get_xaxis().set_ticklabels([\"%d%%\" % (i * 10) for i in range(11)])\n",
    "\n",
    "_ = ax.set(xlabel=\"cumulative percentage of simulations with at least this change\", ylabel=\"gain or loss\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Identifying the 5% value-at-risk"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "percentage_var = 0.05\n",
    "\n",
    "simulated_values[int(len(simulated_values) * percentage_var)]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Visualizing random walks by retaining history"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def simulate_with_history(seed, pf, params, days):\n",
    "    from random import Random\n",
    "    prng = Random()\n",
    "    prng.seed(seed)\n",
    "    pf = pf.copy()\n",
    "    values = [portfolio_value(pf)]\n",
    "    \n",
    "    for day in range(days):\n",
    "        pf = simstep(pf, params, prng)\n",
    "        values.append(portfolio_value(pf))\n",
    "    return values"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Collecting results at each decile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "simulated_results.sort()\n",
    "\n",
    "eleven_results = [simulated_results[int((len(simulated_results) - 1) * i / 10)] for i in range(11)]\n",
    "eleven_seeds = sc.parallelize([seed for (_, seed) in eleven_results])\n",
    "walks = eleven_seeds.map(lambda s: simulate_with_history(s, bpf.value, bparams.value, 5))\n",
    "\n",
    "walk_results = walks.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for c in walk_results:\n",
    "    ax = sns.tsplot(c)\n",
    "\n",
    "_ = ax.set(xlabel=\"day of simulation\", ylabel=\"total portfolio value\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Getting more realistic results\n",
    "\n",
    "Of course, most real-world stock returns aren't normally distributed.  To make a more interesting simulation, we can try to find a distribution that better models the returns we've observed.  We'll start by looking at the actual distributions of returns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rdist = ddf.filter(ddf[\"ticker\"] == \"RHT\").select(\"change\").rdd.map(lambda r: r[\"change\"]).filter(lambda c: c is not None).collect()\n",
    "sns.distplot(rdist)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stock price changes aren't normally-distributed"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from scipy import stats\n",
    "sns.distplot(rdist, fit=stats.norm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Trying different distributions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "symbols = ddf.select(\"ticker\").distinct().sample(True, 0.004).rdd.map(lambda l: l[\"ticker\"]).collect()\n",
    "dfs = ddf.filter(ddf[\"ticker\"].isin(symbols)).select(\"ticker\", \"change\").dropna()\n",
    "sampled_returns = dfs.toPandas()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g = sns.FacetGrid(sampled_returns, row=\"ticker\", sharex=False, sharey=False, aspect=3)\n",
    "_ = g.map(sns.distplot, \"change\", fit=stats.norm)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "g2 = sns.FacetGrid(sampled_returns, row=\"ticker\", sharex=False, sharey=False, aspect=3)\n",
    "_ = g2.map(sns.distplot, \"change\", fit=stats.t)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}