2.3.1 - Content - StreamingPrices - With all events.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Refinitiv Data Platform Library for Python\n",
    "## Content - Streaming Prices - How to use streaming events"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This Notebook demonstrates how to retrieve level 1 streaming data (such as trades and quotes) from the Refinitiv Data Platform. The example will show how to define a StreamingPrices object with registered event handlers so that your application is notified when new data is coming in.\n",
    "\n",
    "Using StreamingPrices that way allows your application to be updated in real-time when data changes on the market. With this event-driven mode, your application can still benefit from the StreamingPrices data cache and use the get_snapshot function to pull out real-time snapshots as Pandas DataFrames.\n",
    "\n",
    "#### Learn more\n",
    "\n",
    "To learn more about the Refinitiv Data Platform Data Libraries just connect to the Refinitiv Developer Community. By [registering](https://developers.refinitiv.com/iam/register) and [login](https://developers.refinitiv.com/iam/login) to the Refinitiv Developer Community portal you will get free access to a number of learning materials like [Quick Start guides](https://developers.refinitiv.com/refinitiv-data-platform/refinitiv-data-platform-libraries/quick-start), [Tutorials](https://developers.refinitiv.com/refinitiv-data-platform/refinitiv-data-platform-libraries/learning), [Documentation](https://developers.refinitiv.com/refinitiv-data-platform/refinitiv-data-platform-libraries/docs) and much more.  \n",
    "\n",
    "#### Getting Help and Support\n",
    "\n",
    "If you have any questions regarding the API usage, please post them on the [Refinitiv Data Platform Q&A Forum](https://community.developers.thomsonreuters.com/spaces/321/index.html). The Refinitiv Developer Community will be very pleased to help you. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Introduction to streaming events\n",
    "\n",
    "Using a StreamingPrices object with events requires you to define event handlers that are called by a background execution thread when new events are received for the instruments you requested. There are 4 different types of event handlers you can optionnaly define depending on the type to events your are interrested in.  \n",
    "\n",
    "### The 4 event types and their related event handlers:\n",
    " - **Refresh events:** Refresh events happen when all fields of one the requested instruments are received. This complete list of fields is sometimes called the 'image' of the instrument. This image that comes with Refresh messages can be later updated by subsequent Update events. When several Refresh events are received for the same instrument, the fields transported by the latest Refresh are considered as the new image. Fields received in previous Refresh events or Update events must be discarded. StreamingPrices objects automatically manage this logic for their internal cache, meaning that when you call get_snapshot you always get the latest and relevant field values for the requested instruments. \n",
    " \n",
    " Refresh event handlers take 3 parameters: \n",
    "  - The StreamingPrices object that received the event\n",
    "  - The name of the concerned instrument\n",
    "  - The fields and values of the Image\n",
    " \n",
    " \n",
    " - **Update events:** Update events are received when fields of a requested instrument change. Update events only contain the fields and values that changed. When the application receives an Update it must update its internal representation of the instrument (if any) accordingly. StreamingPrices objects automatically manage this logic for their internal cache, meaning that when you call get_snapshot you always get the latest values of the requested instruments. \n",
    " \n",
    " Update event handlers take 3 parameters \n",
    "  - The StreamingPrices object that received the event\n",
    "  - The name of the concerned instrument\n",
    "  - The updated fields with their new values\n",
    " \n",
    " \n",
    " - **Status events:** Status events are received when the status of one of the requested instruments changes.\n",
    " \n",
    " Status event handlers take 3 parameters \n",
    "  - The StreamingPrices object that received the event\n",
    "  - The name of the concerned instrument\n",
    "  - The new status of the instrument\n",
    " \n",
    " \n",
    " - **Complete events:** A Complete event is received once all the requested instruments received either a Refresh or a Status event. The Complete event indicates that the StreamingPrices object is complete and that it's internal cache contains the full data set (instruments and fields) that were requested.     \n",
    "\n",
    " Complete event handlers take one parameter\n",
    "  - The StreamingPrices object that is complete\n",
    " \n",
    "**Side note:** As Refresh events and Update events use handlers with the same signature, the same handler can be used for these 2 event types if you do not need to distinguish them \n",
    " \n",
    "### Typical event flow\n",
    "\n",
    "As an example, if you use a StreamingPrices object with event for the following instruments and fields: \n",
    " - Instruments: 'CAD=','GBP=', 'JPY=', 'JUNK'\n",
    " - Fields: 'CF_BID','CF_ASK','OPEN_PRC', 'CF_HIGH','CF_LOW', 'CF_CLOSE'\n",
    "\n",
    "You may receive a flow of events like this one:\n",
    " 1. **Refresh** event for GBP=\n",
    " 1. **Refresh** event for CAD=\n",
    " 1. **Status** event for JUNK\n",
    " 1. **Refresh** event for JPY=\n",
    " 1. **Complete** event => Indicating that data (or status) has been received for all requested instruments\n",
    " 1. **Update** event for JPY=\n",
    " 1. **Update** event for CAD=\n",
    " 1. **Refresh** event for CAD=\n",
    " 1. **Update** event for GBP=\n",
    " 1. **Update** event for CAD=\n",
    " 1. **Update** event for JPY=\n",
    " 1. ...\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Import the library and load credentials\n",
    "\n",
    "Credentials used by this notebook are stored in the ./credentials.ipynb. Please edit ./credentials.ipynb to set your credentials and run the next cell to continue with this scripts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import refinitiv.dataplatform as rdp\n",
    "import datetime\n",
    "\n",
    "%run ./credentials.ipynb"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Open the session of your choice\n",
    "\n",
    "Depending on the access point your application uses to connect to the Refinitiv Data Platform, it needs to call one of the following functions to create and open a session to the platform. This session will become the default session used by all other function calls."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Either\n",
    "\n",
    "Create and open a Desktop session to connect to the Refinitiv Data Platform pvia Eikon 4 or the Refinitiv Workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<refinitiv.dataplatform.core.session.desktop_session.DesktopSession at 0xb0927f0>"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdp.open_desktop_session(APP_KEY)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### or\n",
    "\n",
    "Create and open a Platform session to connect directly to the Refinitiv Data Platform. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<refinitiv.dataplatform.core.session.platform_session.PlatformSession at 0xb0652e8>"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdp.open_platform_session(\n",
    "    APP_KEY, \n",
    "    rdp.GrantPassword(\n",
    "        username = RDP_LOGIN, \n",
    "        password = RDP_PASSWORD\n",
    "    )\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### or\n",
    "\n",
    "Create and open a Deployed Platform session to connect directly to a deployed enterprise platform (a.k.a.TREP). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<refinitiv.dataplatform.core.session.deployed_platform_session.DeployedPlatformSession at 0xadfd748>"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rdp.open_deployed_platform_session(\n",
    "    APP_KEY,\n",
    "    DEPLOYED_PLATFORM_HOST,\n",
    "    DEPLOYED_PLATFORM_USER_NAME\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define callbacks to capture incoming events"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following function will be used to capture Refresh events. It displays the name of the refreshed instrument and its full image (complete list of requested fields)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def display_refreshed_fields(streaming_price, instrument_name, fields):\n",
    "    current_time = datetime.datetime.now().time()\n",
    "    print(current_time, \"- Refresh received for\", instrument_name, \":\", fields)    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following function will be used to capture Update events. It displays the name of the updated instrument and the updated fields."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def display_updated_fields(streaming_price, instrument_name, fields):\n",
    "    current_time = datetime.datetime.now().time()\n",
    "    print(current_time, \"- Update received for\", instrument_name, \":\", fields)    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following function will be used to capture Status events. It displays the name of the instrument and the received status."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def display_status(streaming_price, instrument_name, status):\n",
    "    current_time = datetime.datetime.now().time()\n",
    "    print(current_time, \"- Status received for\", instrument_name, \":\", status)    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following function will be used to capture Complete events. It calls get_snapshot to pull out the memory cache of the StreamingPrice as a Pandas DataFrame and displays it. As a result, the latest values of all requested fields and instruments are displayed in a table. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "def display_complete_snapshot(streaming_prices):\n",
    "    current_time = datetime.datetime.now().time()\n",
    "    print(current_time, \"- StreamingPrice is complete. Full snapshot:\")\n",
    "    display(streaming_prices.get_snapshot())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create StreamingPrices and register the  event callbacks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "streaming_prices = rdp.StreamingPrices(\n",
    "    universe = ['EUR=', 'GBP=', 'JPY=', 'CAD='], \n",
    "    fields   = ['BID', 'ASK', 'OPEN_PRC', 'HST_CLOSE'],\n",
    "    on_refresh = lambda streaming_price, instrument_name, fields : \n",
    "        display_refreshed_fields(streaming_price, instrument_name, fields),\n",
    "    on_update = lambda streaming_price, instrument_name, fields : \n",
    "        display_updated_fields(streaming_price, instrument_name, fields),\n",
    "    on_status = lambda streaming_price, instrument_name, status : \n",
    "        display_status(streaming_price, instrument_name, status),\n",
    "    on_complete = lambda streaming_price : \n",
    "        display_complete_snapshot(streaming_price)\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Open StreamingPrices and wait for events"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "14:33:48.271306 - Status received for GBP= : {'status': <StreamState.Open: 3>, 'code': 'Open', 'message': 'All is well'}\n",
      "14:33:48.271306 - Refresh received for GBP= : {'OPEN_PRC': 1.3077, 'HST_CLOSE': 1.3076, 'BID': 1.3044, 'ASK': 1.3048}\n",
      "14:33:48.271306 - Status received for EUR= : {'status': <StreamState.Open: 3>, 'code': 'Open', 'message': 'All is well'}\n",
      "14:33:48.271306 - Refresh received for EUR= : {'OPEN_PRC': 1.1111, 'HST_CLOSE': 1.1111, 'BID': 1.1121, 'ASK': 1.1125}\n",
      "14:33:48.271306 - Update received for GBP= : {'BID': 1.3044, 'ASK': 1.3048}\n",
      "14:33:48.271306 - Status received for JPY= : {'status': <StreamState.Open: 3>, 'code': 'Open', 'message': 'All is well'}\n",
      "14:33:48.271306 - Refresh received for JPY= : {'OPEN_PRC': 109.54, 'HST_CLOSE': 109.53, 'BID': 109.44, 'ASK': 109.47}\n",
      "14:33:48.271306 - Status received for CAD= : {'status': <StreamState.Open: 3>, 'code': 'Open', 'message': 'All is well'}\n",
      "14:33:48.272306 - Refresh received for CAD= : {'OPEN_PRC': 1.311, 'HST_CLOSE': 1.3113, 'BID': 1.3116, 'ASK': 1.312}\n",
      "14:33:48.272306 - StreamingPrice is complete. Full snapshot:\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>Instrument</th>\n",
       "      <th>OPEN_PRC</th>\n",
       "      <th>HST_CLOSE</th>\n",
       "      <th>BID</th>\n",
       "      <th>ASK</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>EUR=</td>\n",
       "      <td>1.1111</td>\n",
       "      <td>1.1111</td>\n",
       "      <td>1.1121</td>\n",
       "      <td>1.1125</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>GBP=</td>\n",
       "      <td>1.3077</td>\n",
       "      <td>1.3076</td>\n",
       "      <td>1.3044</td>\n",
       "      <td>1.3048</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>JPY=</td>\n",
       "      <td>109.5400</td>\n",
       "      <td>109.5300</td>\n",
       "      <td>109.4400</td>\n",
       "      <td>109.4700</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>CAD=</td>\n",
       "      <td>1.3110</td>\n",
       "      <td>1.3113</td>\n",
       "      <td>1.3116</td>\n",
       "      <td>1.3120</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "  Instrument  OPEN_PRC  HST_CLOSE       BID       ASK\n",
       "0       EUR=    1.1111     1.1111    1.1121    1.1125\n",
       "1       GBP=    1.3077     1.3076    1.3044    1.3048\n",
       "2       JPY=  109.5400   109.5300  109.4400  109.4700\n",
       "3       CAD=    1.3110     1.3113    1.3116    1.3120"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "<StreamState.Open: 3>"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "14:33:48.602238 - Update received for GBP= : {'BID': 1.3045, 'ASK': 1.3046}\n",
      "14:33:48.602238 - Update received for CAD= : {'BID': 1.3119, 'ASK': 1.3123}\n",
      "14:33:49.353081 - Update received for EUR= : {'BID': 1.1121, 'ASK': 1.1123}\n",
      "14:33:49.670018 - Update received for GBP= : {'BID': 1.3043, 'ASK': 1.3048}\n",
      "14:33:49.671017 - Update received for GBP= : {'BID': 1.3043, 'ASK': 1.3048}\n",
      "14:33:49.991953 - Update received for GBP= : {'BID': 1.3045, 'ASK': 1.3048}\n",
      "14:33:50.306890 - Update received for CAD= : {'BID': 1.3119, 'ASK': 1.3124}\n",
      "14:33:50.306890 - Update received for CAD= : {'BID': 1.3116, 'ASK': 1.312}\n",
      "14:33:50.626826 - Update received for GBP= : {'BID': 1.3044, 'ASK': 1.3048}\n",
      "14:33:50.626826 - Update received for GBP= : {'BID': 1.3045, 'ASK': 1.3046}\n",
      "14:33:50.947762 - Update received for EUR= : {'BID': 1.1122, 'ASK': 1.1123}\n",
      "14:33:51.266698 - Update received for CAD= : {'BID': 1.3116, 'ASK': 1.312}\n"
     ]
    }
   ],
   "source": [
    "streaming_prices.open()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Close StreamingPrices when done"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "streaming_prices.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}