pdblp.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "#This is a 3rd Party Implementation of BDH/BDP API Calls\n",
    "#Using this to pull data and handle errors in main code\n",
    "#Must be executed before running the \n",
    "\n",
    "import blpapi\n",
    "import logging\n",
    "import pandas as pd\n",
    "import contextlib\n",
    "import time\n",
    "import numpy as np\n",
    "from datetime import datetime, timedelta\n",
    "from collections import defaultdict\n",
    "from pandas import DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": [
    "@contextlib.contextmanager\n",
    "def bopen(**kwargs):\n",
    "    \"\"\"\n",
    "    Open and manage a BCon wrapper to a Bloomberg API session\n",
    "    Parameters\n",
    "    \n",
    "    **kwargs:\n",
    "        Keyword arguments passed into pdblp.BCon initialization\n",
    "    \"\"\"\n",
    "    con = BCon(**kwargs)\n",
    "    con.start()\n",
    "    try:\n",
    "        yield con\n",
    "    finally:\n",
    "        con.stop()\n",
    "\n",
    "\n",
    "class BCon(object):\n",
    "    def __init__(self, host='localhost', port=8194, debug=False, timeout=5000):\n",
    "        \"\"\"\n",
    "        Create an object which manages connection to the Bloomberg API session\n",
    "        Parameters\n",
    "        ----------\n",
    "        host: str\n",
    "            Host name\n",
    "        port: int\n",
    "            Port to connect to\n",
    "        debug: Boolean {True, False}\n",
    "            Boolean corresponding to whether to log Bloomberg Open API request\n",
    "            and response messages to stdout\n",
    "        timeout: int\n",
    "            Number of milliseconds before timeout occurs when parsing response.\n",
    "            See blp.Session.nextEvent() for more information.\n",
    "        \"\"\"\n",
    "        # Fill SessionOptions\n",
    "        sessionOptions = blpapi.SessionOptions()\n",
    "        sessionOptions.setServerHost(host)\n",
    "        sessionOptions.setServerPort(port)\n",
    "        self._sessionOptions = sessionOptions\n",
    "        # Create a Session\n",
    "        self.session = blpapi.Session(sessionOptions)\n",
    "        self.timeout = timeout\n",
    "        # initialize logger\n",
    "        self.debug = debug\n",
    "\n",
    "    @property\n",
    "    def debug(self):\n",
    "        \"\"\"\n",
    "        When True, print all Bloomberg Open API request and response messages\n",
    "        to stdout\n",
    "        \"\"\"\n",
    "        return self._debug\n",
    "\n",
    "    @debug.setter\n",
    "    def debug(self, value):\n",
    "        \"\"\"\n",
    "        Set whether logging is True or False\n",
    "        \"\"\"\n",
    "        self._debug = value\n",
    "        root = logging.getLogger()\n",
    "        if self._debug:\n",
    "            # log requests and responses\n",
    "            root.setLevel(logging.DEBUG)\n",
    "        else:\n",
    "            # log only failed connections\n",
    "            root.setLevel(logging.INFO)\n",
    "\n",
    "    def start(self):\n",
    "        \"\"\"\n",
    "        start connection and init service for refData\n",
    "        \"\"\"\n",
    "        # Start a Session\n",
    "        if not self.session.start():\n",
    "            logging.info(\"Failed to start session.\")\n",
    "            raise ConnectionError(\"Could not start a blpapi.session\")\n",
    "        self.session.nextEvent()\n",
    "        # Open service to get historical data from\n",
    "        if not self.session.openService(\"//blp/refdata\"):\n",
    "            logging.info(\"Failed to open //blp/refdata\")\n",
    "            raise ConnectionError(\"Could not open a //blp/refdata service\")\n",
    "        self.session.nextEvent()\n",
    "        # Obtain previously opened service\n",
    "        self.refDataService = self.session.getService(\"//blp/refdata\")\n",
    "        self.session.nextEvent()\n",
    "        # Open BSearch Service\n",
    "        if not self.session.openService(\"//blp/exrsvc\"):\n",
    "            logging.error(\"Failed to open //blp/exrsvc\")\n",
    "            raise ConnectionError(\"Could not open a //blp/exrsvc service\")\n",
    "        # Obtain previously opened service\n",
    "        self.exrService = self.session.getService(\"//blp/exrsvc\")\n",
    "        self.session.nextEvent()\n",
    "\n",
    "    def restart(self):\n",
    "        \"\"\"\n",
    "        Restart the blp session\n",
    "        \"\"\"\n",
    "        # Recreate a Session\n",
    "        self.session = blpapi.Session(self._sessionOptions)\n",
    "        self.start()\n",
    "\n",
    "    def _create_req(self, rtype, tickers, flds, ovrds, setvals):\n",
    "        # flush event queue in case previous call errored out\n",
    "        while(self.session.tryNextEvent()):\n",
    "            pass\n",
    "\n",
    "        request = self.refDataService.createRequest(rtype)\n",
    "        for t in tickers:\n",
    "            request.getElement(\"securities\").appendValue(t)\n",
    "        for f in flds:\n",
    "            request.getElement(\"fields\").appendValue(f)\n",
    "        for name, val in setvals:\n",
    "            request.set(name, val)\n",
    "\n",
    "        overrides = request.getElement(\"overrides\")\n",
    "        for ovrd_fld, ovrd_val in ovrds:\n",
    "            ovrd = overrides.appendElement()\n",
    "            ovrd.setElement(\"fieldId\", ovrd_fld)\n",
    "            ovrd.setElement(\"value\", ovrd_val)\n",
    "\n",
    "        return request\n",
    "\n",
    "    def bdh(self, tickers, flds, start_date, end_date, elms=[],\n",
    "            ovrds=[], longdata=False):\n",
    "        \"\"\"\n",
    "        Get tickers and fields, return pandas Dataframe with columns as\n",
    "        MultiIndex with levels \"ticker\" and \"field\" and indexed by \"date\".\n",
    "        If long data is requested return DataFrame with columns\n",
    "        [\"date\", \"ticker\", \"field\", \"value\"].\n",
    "        Parameters\n",
    "        ----------\n",
    "        tickers: {list, string}\n",
    "            String or list of strings corresponding to tickers\n",
    "        flds: {list, string}\n",
    "            String or list of strings corresponding to FLDS\n",
    "        start_date: string\n",
    "            String in format YYYYmmdd\n",
    "        end_date: string\n",
    "            String in format YYYYmmdd\n",
    "        elms: list of tuples\n",
    "            List of tuples where each tuple corresponds to the other elements\n",
    "            to be set, e.g. [(\"periodicityAdjustment\", \"ACTUAL\")]\n",
    "            Refer to A.2.4 HistoricalDataRequest in the Developers Guide for\n",
    "            more info on these values\n",
    "        ovrds: list of tuples\n",
    "            List of tuples where each tuple corresponds to the override\n",
    "            field and value\n",
    "        longdata: boolean\n",
    "            Whether data should be returned in long data format or pivoted\n",
    "        \"\"\"\n",
    "\n",
    "        elms = list(elms)\n",
    "\n",
    "        data = self._bdh_list(tickers, flds, start_date, end_date,\n",
    "                              elms, ovrds)\n",
    "\n",
    "        df = DataFrame(data)\n",
    "        df.columns = [\"date\", \"ticker\", \"field\", \"value\"]\n",
    "        df.loc[:, \"date\"] = pd.to_datetime(df.loc[:, \"date\"])\n",
    "        if not longdata:\n",
    "            cols = ['ticker', 'field']\n",
    "            df = df.set_index(['date'] + cols).unstack(cols)\n",
    "            df.columns = df.columns.droplevel(0)\n",
    "\n",
    "        return df\n",
    "\n",
    "    def _bdh_list(self, tickers, flds, start_date, end_date, elms,\n",
    "                  ovrds):\n",
    "\n",
    "        if type(tickers) is not list:\n",
    "            tickers = [tickers]\n",
    "        if type(flds) is not list:\n",
    "            flds = [flds]\n",
    "\n",
    "        setvals = elms\n",
    "        setvals.append((\"startDate\", start_date))\n",
    "        setvals.append((\"endDate\", end_date))\n",
    "\n",
    "        request = self._create_req(\"HistoricalDataRequest\", tickers, flds,\n",
    "                                   ovrds, setvals)\n",
    "\n",
    "        logging.debug(\"Sending Request:\\n %s\" % request)\n",
    "        # Send the request\n",
    "        self.session.sendRequest(request)\n",
    "        data = []\n",
    "        # Process received events\n",
    "        while(True):\n",
    "            # We provide timeout to give the chance for Ctrl+C handling:\n",
    "            ev = self.session.nextEvent(self.timeout)\n",
    "            for msg in ev:\n",
    "                logging.debug(\"Message Received:\\n %s\" % msg)\n",
    "                has_security_error = (msg.getElement('securityData')\n",
    "                                      .hasElement('securityError'))\n",
    "                has_field_exception = (msg.getElement('securityData')\n",
    "                                       .getElement(\"fieldExceptions\")\n",
    "                                       .numValues() > 0)\n",
    "                if has_security_error or has_field_exception:\n",
    "                    raise ValueError(msg)\n",
    "                ticker = (msg.getElement('securityData')\n",
    "                          .getElement('security').getValue())\n",
    "                fldDatas = (msg.getElement('securityData')\n",
    "                            .getElement('fieldData'))\n",
    "                for fd in fldDatas.values():\n",
    "                    dt = fd.getElement('date').getValue()\n",
    "                    for element in fd.elements():\n",
    "                        fname = str(element.name())\n",
    "                        if fname == \"date\":\n",
    "                            continue\n",
    "                        val = element.getValue()\n",
    "                        data.append((dt, ticker, fname, val))\n",
    "            if ev.eventType() == blpapi.Event.RESPONSE:\n",
    "                # Response completely received, so we could exit\n",
    "                break\n",
    "\n",
    "        return data\n",
    "\n",
    "    def ref(self, tickers, flds, ovrds=[]):\n",
    "        \"\"\"\n",
    "        Make a reference data request, get tickers and fields, return long\n",
    "        pandas Dataframe with columns [ticker, field, value]\n",
    "        Parameters\n",
    "        ----------\n",
    "        tickers: {list, string}\n",
    "            String or list of strings corresponding to tickers\n",
    "        flds: {list, string}\n",
    "            String or list of strings corresponding to FLDS\n",
    "        ovrds: list of tuples\n",
    "            List of tuples where each tuple corresponds to the override\n",
    "            field and value\n",
    "        \"\"\"\n",
    "\n",
    "        if type(tickers) is not list:\n",
    "            tickers = [tickers]\n",
    "        if type(flds) is not list:\n",
    "            flds = [flds]\n",
    "\n",
    "        request = self._create_req(\"ReferenceDataRequest\", tickers, flds,\n",
    "                                   ovrds, [])\n",
    "\n",
    "        logging.debug(\"Sending Request:\\n %s\" % request)\n",
    "        self.session.sendRequest(request)\n",
    "        data = self._parse_ref(flds)\n",
    "        data = DataFrame(data)\n",
    "        data.columns = [\"ticker\", \"field\", \"value\"]\n",
    "        return data\n",
    "\n",
    "    def _parse_ref(self, flds, keep_corrId=False, sent_events=1):\n",
    "        data = []\n",
    "        # Process received events\n",
    "        while(True):\n",
    "            # We provide timeout to give the chance for Ctrl+C handling:\n",
    "            ev = self.session.nextEvent(self.timeout)\n",
    "            for msg in ev:\n",
    "                logging.debug(\"Message Received:\\n %s\" % msg)\n",
    "                if keep_corrId:\n",
    "                    corrId = [msg.correlationIds()[0].value()]\n",
    "                else:\n",
    "                    corrId = []\n",
    "                secDataArray = msg.getElement('securityData')\n",
    "                for secDataElm in secDataArray.values():\n",
    "                    ticker = secDataElm.getElement(\"security\").getValue()\n",
    "                    if secDataElm.hasElement(\"securityError\"):\n",
    "                        raise ValueError(\"Unknow security %s\" % ticker)\n",
    "                    self._check_fieldExceptions(secDataElm.getElement(\"fieldExceptions\"))  # NOQA\n",
    "                    fieldData = secDataElm.getElement('fieldData')\n",
    "                    for j in range(len(flds)):\n",
    "                        fld = flds[j]\n",
    "                        # this is a slight hack but if a fieldData response\n",
    "                        # does not have the element fld and this is not a bad\n",
    "                        # field (which is checked above) then the assumption is\n",
    "                        # that this is a not applicable field, thus set NaN\n",
    "                        if not fieldData.hasElement(fld):\n",
    "                            dataj = [ticker, fld, pd.np.NaN]\n",
    "                            dataj.extend(corrId)\n",
    "                            data.append(dataj)\n",
    "                        # this is for dealing with requests which return arrays\n",
    "                        # of values for a single field\n",
    "                        elif fieldData.getElement(fld).isArray():\n",
    "                            for field in fieldData.getElement(fld).values():\n",
    "                                # if the elements of the array have multiple\n",
    "                                # subelements this will just append them all\n",
    "                                # into a list\n",
    "                                for elm in field.elements():\n",
    "                                    mfld = fld + \":\" + str(elm.name())\n",
    "                                    if not elm.isNull():\n",
    "                                        val = elm.getValue()\n",
    "                                    else:\n",
    "                                        val = pd.np.NaN\n",
    "                                    dataj = [ticker, mfld, val]\n",
    "                                    dataj.extend(corrId)\n",
    "                                    data.append(dataj)\n",
    "                        else:\n",
    "                            val = fieldData.getElement(fld).getValue()\n",
    "                            dataj = [ticker, fld, val]\n",
    "                            dataj.extend(corrId)\n",
    "                            data.append(dataj)\n",
    "\n",
    "            if ev.eventType() == blpapi.Event.RESPONSE:\n",
    "                # Response completely received, so we could exit\n",
    "                sent_events = sent_events - 1\n",
    "                if sent_events == 0:\n",
    "                    break\n",
    "            # calls can occassionally timeout in unpredictable manner due\n",
    "            # to variability in BBG response times\n",
    "            elif ev.eventType() == blpapi.Event.TIMEOUT:\n",
    "                raise RuntimeError(\"Timeout, increase BCon.timeout attribute\")\n",
    "\n",
    "        return data\n",
    "\n",
    "    @staticmethod\n",
    "    def _check_fieldExceptions(fieldExceptions):\n",
    "        # iterate over an array of fieldExceptions and check for a\n",
    "        # INVALID_FIELD error\n",
    "\n",
    "        INVALID_FIELD = 'INVALID_FIELD'\n",
    "        for fe in fieldExceptions.values():\n",
    "            category = fe.getElement(\"errorInfo\").getElement(\"subcategory\").getValue()  # NOQA\n",
    "            if category == INVALID_FIELD:\n",
    "                raise ValueError(\"%s: %s\" % (fe.getElement(\"fieldId\").getValue(), category))  # NOQA\n",
    "\n",
    "    def ref_hist(self, tickers, flds, dates, ovrds=[],\n",
    "                 date_field=\"REFERENCE_DATE\"):\n",
    "        \"\"\"\n",
    "        Get tickers and fields, periodically override date_field to create\n",
    "        a time series. Return pandas dataframe with column MultiIndex\n",
    "        of tickers and fields if multiple fields given, Index otherwise.\n",
    "        If single field is given DataFrame is ordered same as tickers,\n",
    "        otherwise MultiIndex is sorted\n",
    "        Parameters\n",
    "        ----------\n",
    "        tickers: {list, string}\n",
    "            String or list of strings corresponding to tickers\n",
    "        flds: {list, string}\n",
    "            String or list of strings corresponding to FLDS\n",
    "        dates: list\n",
    "            list of date strings in the format YYYYmmdd\n",
    "        ovrds: list of tuples\n",
    "            List of tuples where each tuple corresponds to the override\n",
    "            field and value. This should not include the date_field which will\n",
    "            be iteratively overridden\n",
    "        date_field: str\n",
    "            Field to iteratively override for requesting historical data,\n",
    "            e.g. REFERENCE_DATE, CURVE_DATE, etc.\n",
    "        \"\"\"\n",
    "        # correlationIDs should be unique to a session so rather than\n",
    "        # managing unique IDs for the duration of the session just restart\n",
    "        # a session for each call\n",
    "        self.restart()\n",
    "        if type(tickers) is not list:\n",
    "            tickers = [tickers]\n",
    "        if type(flds) is not list:\n",
    "            flds = [flds]\n",
    "\n",
    "        request = self._create_req(\"ReferenceDataRequest\", tickers, flds,\n",
    "                                   ovrds, [])\n",
    "\n",
    "        overrides = request.getElement(\"overrides\")\n",
    "        if len(dates) == 0:\n",
    "            raise ValueError(\"dates must by non empty\")\n",
    "        ovrd = overrides.appendElement()\n",
    "        for dt in dates:\n",
    "            ovrd.setElement(\"fieldId\", date_field)\n",
    "            ovrd.setElement(\"value\", dt)\n",
    "            # CorrelationID used to keep track of which response coincides with\n",
    "            # which request\n",
    "            cid = blpapi.CorrelationId(dt)\n",
    "            logging.debug(\"Sending Request:\\n %s\" % request)\n",
    "            self.session.sendRequest(request, correlationId=cid)\n",
    "\n",
    "        data = self._parse_ref(flds, keep_corrId=True, sent_events=len(dates))\n",
    "        data = pd.DataFrame(data)\n",
    "        data.columns = ['ticker', 'field', 'value', 'date']\n",
    "        data = data.sort_values(by='date')\n",
    "        data = data.reset_index(drop=True)\n",
    "        data = data.loc[:, ['date', 'field', 'ticker', 'value']]\n",
    "        return data\n",
    "\n",
    "    def bdib(self, ticker, start_datetime, end_datetime, event_type, interval,\n",
    "             elms=[]):\n",
    "        \"\"\"\n",
    "        Get Open, High, Low, Close, Volume, and numEvents for a ticker.\n",
    "        Return pandas dataframe\n",
    "        Parameters\n",
    "        ----------\n",
    "        ticker: string\n",
    "            String corresponding to ticker\n",
    "        start_datetime: string\n",
    "            UTC datetime in format YYYY-mm-ddTHH:MM:SS\n",
    "        end_datetime: string\n",
    "            UTC datetime in format YYYY-mm-ddTHH:MM:SS\n",
    "        event_type: string {TRADE, BID, ASK, BID_BEST, ASK_BEST, BEST_BID,\n",
    "                           BEST_ASK}\n",
    "            Requested data event type\n",
    "        interval: int {1... 1440}\n",
    "            Length of time bars\n",
    "        elms: list of tuples\n",
    "            List of tuples where each tuple corresponds to the other elements\n",
    "            to be set, refer to A.2.8 IntradayBarRequest in the\n",
    "            Developers Guide for more info on these values\n",
    "        \"\"\"\n",
    "        # flush event queue in case previous call errored out\n",
    "        while(self.session.tryNextEvent()):\n",
    "            pass\n",
    "\n",
    "        # Create and fill the request for the historical data\n",
    "        request = self.refDataService.createRequest(\"IntradayBarRequest\")\n",
    "        request.set(\"security\", ticker)\n",
    "        request.set(\"eventType\", event_type)\n",
    "        request.set(\"interval\", interval)  # bar interval in minutes\n",
    "        request.set(\"startDateTime\", start_datetime)\n",
    "        request.set(\"endDateTime\", end_datetime)\n",
    "        for name, val in elms:\n",
    "            request.set(name, val)\n",
    "\n",
    "        logging.debug(\"Sending Request:\\n %s\" % request)\n",
    "        # Send the request\n",
    "        self.session.sendRequest(request)\n",
    "        # defaultdict - later convert to pandas\n",
    "        data = defaultdict(dict)\n",
    "        # Process received events\n",
    "        flds = ['open', 'high', 'low', 'close', 'volume', 'numEvents']\n",
    "        while(True):\n",
    "            # We provide timeout to give the chance for Ctrl+C handling:\n",
    "            ev = self.session.nextEvent(self.timeout)\n",
    "            for msg in ev:\n",
    "                logging.debug(\"Message Received:\\n %s\" % msg)\n",
    "                barTick = (msg.getElement('barData')\n",
    "                           .getElement('barTickData'))\n",
    "                for i in range(barTick.numValues()):\n",
    "                    for fld in flds:\n",
    "                        dt = barTick.getValue(i).getElement(0).getValue()\n",
    "                        val = (barTick.getValue(i).getElement(fld).getValue())\n",
    "                        data[(fld)][dt] = val\n",
    "\n",
    "            if ev.eventType() == blpapi.Event.RESPONSE:\n",
    "                # Response completly received, so we could exit\n",
    "                break\n",
    "        data = DataFrame(data)\n",
    "        if not data.empty:\n",
    "            data.index = pd.to_datetime(data.index)\n",
    "            data = data[flds]\n",
    "        return data\n",
    "\n",
    "    def bsrch(self, domain):\n",
    "        \"\"\"This function uses the Bloomberg API to retrieve 'bsrch'\n",
    "        (Bloomberg SRCH Data) queries. Returns list of tickers.\n",
    "        Parameters\n",
    "        ----------\n",
    "        domain: string\n",
    "        A character string with the name of the domain to execute.\n",
    "        It can be a user defined SRCH screen, commodity screen or\n",
    "        one of the variety of Bloomberg examples. All domains are in the format\n",
    "        <domain>:<search_name>. Example \"COMDTY:NGFLOW\"\n",
    "        Returns\n",
    "        -------\n",
    "        data: pandas.DataFrame\n",
    "        List of bloomberg tickers from the BSRCH\n",
    "        \"\"\"\n",
    "        request = self.exrService.createRequest(\"ExcelGetGridRequest\")\n",
    "        request.set(\"Domain\", domain)\n",
    "        self.session.sendRequest(request)\n",
    "        data = []\n",
    "        # Process received events\n",
    "        while True:\n",
    "            # We provide timeout to give the chance for Ctrl+C handling:\n",
    "            event = self.session.nextEvent(self.timeout)\n",
    "            if event.eventType() == blpapi.Event.RESPONSE or \\\n",
    "               event.eventType() == blpapi.Event.PARTIAL_RESPONSE:\n",
    "                for msg in event:\n",
    "                    logging.debug(msg)\n",
    "                    for v in msg.getElement(\"DataRecords\").values():\n",
    "                        for f in v.getElement(\"DataFields\").values():\n",
    "                            data.append(f.getElementAsString(\"StringValue\"))\n",
    "            if event.eventType() == blpapi.Event.RESPONSE:\n",
    "                break\n",
    "        return pd.DataFrame(data)\n",
    "\n",
    "    def stop(self):\n",
    "        \"\"\"\n",
    "        Close the blp session\n",
    "        \"\"\"\n",
    "        self.session.stop()\n",
    "        \n",
    "#***************************************************************************************************\n",
    "#***************************************************************************************************\n",
    "#***************************************************************************************************\n",
    "#***************************************************************************************************\n",
    "#**********************"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "deletable": true,
    "editable": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (sandboxed)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.5.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}