extract_metrics_from_fillings.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XP5N3UtgQ1AU"
      },
      "outputs": [],
      "source": [
        "%pip install sec-api\n",
        "%pip install -U transformers==3.0.0\n",
        "%python -m nltk.downloader punkt\n",
        "%git clone https://github.com/patil-suraj/question_generation.git\n",
        "%cd question_generation"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 2,
      "metadata": {
        "id": "gKY9J8hE8l5i"
      },
      "outputs": [],
      "source": [
        "import spacy\n",
        "\n",
        "from pipelines import pipeline\n",
        "from nltk.tokenize import sent_tokenize\n",
        "import re\n",
        "\n",
        "from sec_api import ExtractorApi"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "id": "tE9wBXkbdwW5"
      },
      "outputs": [],
      "source": [
        "def get_section(filing_url, section) :\n",
        "    res = filing_url[:filing_url.index(\"htm\") + len(\"htm\")]\n",
        "    section_text = extractorApi.get_section(res, section, \"text\")\n",
        "    return section_text"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 4,
      "metadata": {
        "id": "NsvE_vuV8l5j"
      },
      "outputs": [],
      "source": [
        "def preprocess_text(text):\n",
        "    temp_text = text.lower().replace('\\n',' ').replace(' %','%')\n",
        "    return temp_text"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 5,
      "metadata": {
        "id": "5izmPP7cFKx8"
      },
      "outputs": [],
      "source": [
        "def extract_metric_vals(text, val_type=\"PERCENT\", NER=None):\n",
        "    if val_type == \"PERCENT\":\n",
        "        return re.findall(r'(\\d+(?:\\.\\d+)?%?(?!\\S))', text)\n",
        "    if val_type == \"NUMBER\":\n",
        "        return re.findall(r\"[+-]?([0-9]+\\.?[0-9]*|\\.[0-9]+)\", text)\n",
        "    if val_type == \"RATIO\":\n",
        "        return re.findall(r\"([0-9]+:[0-9]+)\", text)\n",
        "    if val_type == \"MONEY\":\n",
        "        values = []\n",
        "        entities = NER(text)\n",
        "        for w in entities.ents:\n",
        "            if w.label_ == 'MONEY':\n",
        "                values.append(w.text)\n",
        "        return values\n",
        "    return []"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 6,
      "metadata": {
        "id": "slnYQPMa8l5j"
      },
      "outputs": [],
      "source": [
        "# 1. search the metric name in the document\n",
        "def search_metric(text_list, metric_list):\n",
        "    matched_indices = []\n",
        "    idx = 0\n",
        "    while idx < len(text_list):\n",
        "        if text_list[idx : idx+len(metric_list)] == metric_list:\n",
        "            matched_indices.append(idx)\n",
        "        idx += 1\n",
        "    return matched_indices\n",
        "\n",
        "# 2. get k words before and after the searched metric\n",
        "def extract_phrases(text_list, matched_indices, k):\n",
        "    phrases_extracted = []\n",
        "    for idx in matched_indices:\n",
        "        phrase = \"\"\n",
        "        for i in range(-k,k+1):\n",
        "            if idx+i < 0 or idx+i > len(text_list)-1:\n",
        "                continue\n",
        "            phrase += text_list[idx+i] + \" \"\n",
        "        phrases_extracted.append(phrase)\n",
        "    return phrases_extracted\n",
        "\n",
        "# 3. apply NER and check for corresonding entity\n",
        "def find_possible_values(text, metric, NER, k, val_type='PERCENT'):\n",
        "    text = text.replace(',', ' ').replace('-', ' ')\n",
        "    metric = metric.replace('-', ' ')\n",
        "    text_list = text.split(' ')\n",
        "    metric_list = metric.split(' ')\n",
        "    matched_indices = search_metric(text_list, metric_list)\n",
        "    phrases_extracted = extract_phrases(text_list, matched_indices, k)\n",
        "    possible_values = []\n",
        "    for phrase in phrases_extracted:\n",
        "        possible_values += extract_metric_vals(phrase, val_type, NER)\n",
        "    return possible_values"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 7,
      "metadata": {
        "id": "NyINcq8vQ_WJ"
      },
      "outputs": [],
      "source": [
        "def filter_passage(doc,metric) :\n",
        "    sents = sent_tokenize(doc)\n",
        "    filtered_sents = \".\".join(s for s in sents if metric in s)\n",
        "    return filtered_sents\n",
        "\n",
        "def get_output(passage,question,metric,NER,val_type='PERCENT') :\n",
        "    tex = preprocess_text(passage)\n",
        "    filtered_passage = filter_passage(tex,metric)\n",
        "    ans = nlp({  \"question\": question,  \"context\": filtered_passage})\n",
        "    ans = ans.replace(',', ' ')\n",
        "    output_values = extract_metric_vals(ans, val_type, NER)\n",
        "    return output_values\n",
        "\n",
        "def get_correct_value(possible_values, output_values):\n",
        "    correct_values = []\n",
        "    for val1 in output_values:\n",
        "        for val2 in possible_values:\n",
        "            if val1 == val2 and val1 not in correct_values:\n",
        "                correct_values.append(val1)    \n",
        "    if len(correct_values) > 0:\n",
        "        return correct_values[-1]\n",
        "    return ''"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "kWd8QjqO81kv"
      },
      "outputs": [],
      "source": [
        "NER = spacy.load(\"en_core_web_sm\", disable=[\"tok2vec\", \"tagger\", \"parser\", \"attribute_ruler\", \"lemmatizer\"])\n",
        "nlp = pipeline(\"multitask-qa-qg\")"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 13,
      "metadata": {
        "id": "_ohvGki1dkt5"
      },
      "outputs": [],
      "source": [
        "api_key = '8aa664eee7896b98038d05ca58f3779d3d9dfa9ca28ad1394702477226556894'\n",
        "extractorApi = ExtractorApi(api_key)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": 23,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "pGxe9wkzAaxF",
        "outputId": "8bf5982c-761c-497a-ea06-97e18a21cf6f"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "0.7%\n"
          ]
        }
      ],
      "source": [
        "metric = \"churn rate\"\n",
        "val_type = \"PERCENT\"\n",
        "k = 6\n",
        "url = \"https://www.sec.gov/Archives/edgar/data/0001710583/000171058319000010/swch12311810-k.htm\"\n",
        "relevant_sections = ['1', '1A', '6', '7']\n",
        "for sec in relevant_sections:\n",
        "    text = get_section(url, sec)\n",
        "    text = preprocess_text(text)\n",
        "    possible_values = find_possible_values(text, metric, NER, k, val_type)\n",
        "    output_values = get_output(text, f'What is the value of {metric}?', metric, NER, val_type)\n",
        "    correct_value = get_correct_value(possible_values, output_values)\n",
        "    if len(correct_value) > 0:\n",
        "        break\n",
        "print(correct_value)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "MSEkT0ygAvf6"
      },
      "outputs": [],
      "source": []
    }
  ],
  "metadata": {
    "colab": {
      "name": "Untitled1.ipynb",
      "provenance": []
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}