08.Inserting-missing-PSC-statements.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Loading the \"missing\" PSC statements\n",
    "\n",
    "Some companies make a statement that the PSC has not yet been identified. Here we build those statement nodes and match the appropriate companies to them."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/ec2-user/anaconda3/lib/python3.6/site-packages/odo/backends/pandas.py:94: FutureWarning: pandas.tslib is deprecated and will be removed in a future version.\n",
      "You can access NaTType as type(pandas.NaT)\n",
      "  @convert.register((pd.Timestamp, pd.Timedelta), (pd.tslib.NaTType, type(None)))\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "import json\n",
    "from pandas.io.json import json_normalize;\n",
    "\n",
    "import blaze as bz"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First load the data and do some tidying up so we don't waste RAM"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "original_psc_data = pd.read_json('../data/psc_snapshot-2017-09-08.json')\n",
    "all_records_psc = pd.concat([original_psc_data['company_number'],json_normalize(original_psc_data['data'])],axis=1)\n",
    "del original_psc_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "False    404603\n",
       "Name: statement, dtype: int64"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "all_records_psc[~all_records_psc.statement.isnull()].statement.map(lambda x: ',' in x).value_counts()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So there are 404,603 statements of a PSC not being found for a company. Below we see the breakdown in the types of statement they can make"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "no-individual-or-entity-with-signficant-control                328286\n",
       "steps-to-find-psc-not-yet-completed                             26806\n",
       "psc-exists-but-not-identified                                   22995\n",
       "psc-details-not-confirmed                                       20025\n",
       "steps-to-find-psc-not-yet-completed-partnership                  3400\n",
       "no-individual-or-entity-with-signficant-control-partnership       836\n",
       "psc-exists-but-not-identified-partnership                         760\n",
       "psc-details-not-confirmed-partnership                             726\n",
       "psc-contacted-but-no-response                                     717\n",
       "restrictions-notice-issued-to-psc                                  48\n",
       "psc-has-failed-to-confirm-changed-details                           4\n",
       "Name: statement, dtype: int64"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "all_records_psc.statement.value_counts()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Inserting the statements into Neo4j\n",
    "\n",
    "Assuming that we have all the nodes up front we can build the family of statements that a company can register. In theory we should only do this once when we are initially building the database. Afterwards, we will have clashes or produce duplicate nodes unless we MERGE rather than CREATE."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from neo4j.v1 import GraphDatabase\n",
    "driver = GraphDatabase.driver(\"bolt://10.0.0.1:7687\", auth=(\"myusername\", \"mypassword\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "with driver.session() as session:\n",
    "    for statement_type in all_records_psc.statement.dropna().unique():\n",
    "        session.run(\"CREATE (s:Statement {type: {statement_type}})\", statement_type=statement_type)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Now to connect companies to statements\n",
    "\n",
    "1. First let's define a function that takes a company record and creates the relationship back to the statement. \n",
    "2. Then we need to loop over all companies that have made statements and insert those into the neo4j database.\n",
    "3. We will chunk over the data so as not to put excess strain on the database."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def write_no_psc_statement(input_data):\n",
    "    \"\"\"Function writes company records where there is no PSC statement to Neo4j database\"\"\"\n",
    "    with driver.session() as session:\n",
    "        session.run((\"UNWIND {list} AS d \"\n",
    "                     \"MERGE (c:Company {uid: d.company_id}) \"\n",
    "                     \"MERGE (s:Statement {type: d.statement}) \"\n",
    "                     \"MERGE (c)-[:DECLARED]->(s);\"), \n",
    "                    {\"list\": input_data})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We shall loop over the data in chunks of 50k records."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/ec2-user/anaconda3/lib/python3.6/site-packages/ipykernel_launcher.py:2: SettingWithCopyWarning: \n",
      "A value is trying to be set on a copy of a slice from a DataFrame.\n",
      "Try using .loc[row_indexer,col_indexer] = value instead\n",
      "\n",
      "See the caveats in the documentation: http://pandas.pydata.org/pandas-docs/stable/indexing.html#indexing-view-versus-copy\n",
      "  \n"
     ]
    }
   ],
   "source": [
    "no_statement_df = all_records_psc[all_records_psc.statement.notnull()]\n",
    "no_statement_df['company_id'] = no_statement_df['links.self'].map(lambda x: x.split('/')[2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(50000, 2)\n",
      "(50000, 2)\n",
      "(50000, 2)\n",
      "(50000, 2)\n",
      "(50000, 2)\n",
      "(50000, 2)\n",
      "(50000, 2)\n",
      "(50000, 2)\n",
      "(4603, 2)\n"
     ]
    }
   ],
   "source": [
    "for chunk in bz.odo(no_statement_df[['company_id', 'statement']], target=bz.chunks(pd.DataFrame), chunksize=50000):\n",
    "    print(chunk.shape)\n",
    "    input_data = [v for k,v in chunk.T.to_dict().items()]\n",
    "    write_no_psc_statement(input_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "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.6.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}