data_sqlserver.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import datetime\n",
    "import os\n",
    "import re\n",
    "from zipfile import ZipFile\n",
    "\n",
    "import pandas as pd\n",
    "\n",
    "import requests\n",
    "from sqlalchemy import create_engine\n",
    "from tqdm import trange\n",
    "\n",
    "from sql_sqlserver import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Config ENV \"dev\" loaded!\n"
     ]
    }
   ],
   "source": [
    "from datalayer_sqlserver import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def Creat(con):\n",
    "    cu = con.cursor()\n",
    "    create_cmds = [\n",
    "        create_cftc_cit_supplement_table_cmd,\n",
    "        create_cftc_futures_table_cmd,\n",
    "        create_cftc_combined_table_cmd,\n",
    "        create_cftc_tff_futures_table_cmd,\n",
    "        create_cftc_tff_combined_table_cmd,\n",
    "        create_cftc_disagg_futures_table_cmd,\n",
    "        create_cftc_disagg_combined_table_cmd\n",
    "    ]\n",
    "    for cmd in create_cmds:\n",
    "        cu.execute(cmd)\n",
    "        con.commit()\n",
    "\n",
    "\n",
    "def Oldest(db):\n",
    "    if db == 'api_cftc_futures':\n",
    "        s = datetime.date(1986, 1, 1)\n",
    "    elif db == 'api_cftc_combined':\n",
    "        s = datetime.date(1995, 1, 1)\n",
    "    else:\n",
    "        s = datetime.date(2006, 1, 1)\n",
    "    return s\n",
    "\n",
    "\n",
    "def DataDownload(y, db):\n",
    "    if db == 'api_cftc_cit_supplement':\n",
    "        url = f\"https://www.cftc.gov/files/dea/history/dea_cit_xls_{str(y)}.zip\"\n",
    "    elif db == 'api_cftc_futures':\n",
    "        if y <= 2003:\n",
    "            url = f\"https://www.cftc.gov/files/dea/history/deafut_xls_{str(y)}.zip\"\n",
    "        else:\n",
    "            url = f\"https://www.cftc.gov/files/dea/history/dea_fut_xls_{str(y)}.zip\"\n",
    "    elif db == 'api_cftc_combined':\n",
    "        if y <= 2003:\n",
    "            url = f'https://www.cftc.gov/files/dea/history/deacom_xls_{str(y)}.zip'\n",
    "        else:\n",
    "            url = f'https://www.cftc.gov/files/dea/history/dea_com_xls_{str(y)}.zip'\n",
    "    elif db == 'api_cftc_tff_futures':\n",
    "        if y <= 2016:\n",
    "            url = \"https://www.cftc.gov/files/dea/history/fin_fut_xls_2006_2016.zip\"\n",
    "        else:\n",
    "            url = f\"https://www.cftc.gov/files/dea/history/fut_fin_xls_{str(y)}.zip\"\n",
    "    elif db == 'api_cftc_tff_combined':\n",
    "        if y <= 2016:\n",
    "            url = 'https://www.cftc.gov/files/dea/history/fin_com_xls_2006_2016.zip'\n",
    "        else:\n",
    "            url = f'https://www.cftc.gov/files/dea/history/com_fin_xls_{str(y)}.zip'\n",
    "    elif db == 'api_cftc_disagg_futures':\n",
    "        if y <= 2015:\n",
    "            url = 'https://www.cftc.gov/files/dea/history/fut_disagg_xls_hist_2006_2016.zip'\n",
    "        else:\n",
    "            url = f'https://www.cftc.gov/files/dea/history/fut_disagg_xls_{str(y)}.zip'\n",
    "    elif db == 'api_cftc_disagg_combined':\n",
    "        if y <= 2015:\n",
    "            url = 'https://www.cftc.gov/files/dea/history/com_disagg_xls_hist_2006_2016.zip'\n",
    "        else:\n",
    "            url = f'https://www.cftc.gov/files/dea/history/com_disagg_xls_{str(y)}.zip'\n",
    "    filename = Download(url)\n",
    "    return filename, os.path.basename(url)\n",
    "\n",
    "\n",
    "def Download(u):\n",
    "    resp = requests.get(u)\n",
    "    file_path = os.path.join(os.path.basename(u))\n",
    "    with open(file_path, 'wb') as file:\n",
    "        file.write(resp.content)\n",
    "    with ZipFile(file_path) as zip_ref:\n",
    "        filename = zip_ref.namelist()[0]\n",
    "        zip_ref.extractall()\n",
    "        if len(zip_ref.namelist()) > 1:\n",
    "            os.remove(zip_ref.namelist()[1])\n",
    "    return filename\n",
    "\n",
    "\n",
    "def Process(p, s):\n",
    "    df = pd.read_excel(p)\n",
    "    df = df.drop(columns=['As_of_Date_In_Form_YYMMDD'])\n",
    "    header = df.columns.tolist()\n",
    "    dict = {}\n",
    "    for h in header:\n",
    "        if re.search(r'Report_Date_as', h):\n",
    "            dict[h] = 'date'\n",
    "        elif h[-1] == ' ':\n",
    "            dict[h] = h[:-1].lower()\n",
    "        elif re.search(r'Spead', h):\n",
    "            dict[h] = (re.sub(r'Spead', 'Spread', h)).lower()\n",
    "        elif re.search(r'__', h):\n",
    "            dict[h] = (re.sub(r'__', '_', h)).lower()\n",
    "        else:\n",
    "            dict[h] = h.lower()\n",
    "    df = df.rename(columns=dict)\n",
    "    df = df[df.date > pd.Timestamp(s)]\n",
    "    df['date'] = df['date'].apply(lambda x: x.strftime(\"%Y-%m-%d\"))\n",
    "    return df\n",
    "\n",
    "\n",
    "def SQL(df, db):\n",
    "    engine = gen_default_sa_engine()\n",
    "    df.to_sql(db, engine, index=False, if_exists='append')\n",
    "\n",
    "\n",
    "def main(tbl, c):\n",
    "    qq = ['api_cftc_tff_futures', 'api_cftc_tff_combined']\n",
    "    gg = ['api_cftc_disagg_futures', 'api_cftc_disagg_combined']\n",
    "    c.execute(f\"SELECT max(date) from {tbl}\")\n",
    "    start = c.fetchone()[0]\n",
    "    if start is None:\n",
    "        start = Oldest(tbl)\n",
    "    end = datetime.datetime.today()\n",
    "    print(f'==={tbl}===')\n",
    "    for i in trange(start.year, end.year + 1):\n",
    "        if tbl in qq and start.year < i <= 2016:\n",
    "            continue\n",
    "        if tbl in gg and start.year < i <= 2015:\n",
    "            continue\n",
    "        path, zpath = DataDownload(i, tbl)\n",
    "        if (not (path)):\n",
    "            print('Error')\n",
    "            continue\n",
    "        cit_df = Process(path, start)\n",
    "        SQL(cit_df, tbl)\n",
    "        os.remove(zpath)\n",
    "        os.remove(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "conn = conn_tds()\n",
    "cur = conn.cursor()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 建表用\n",
    "Creat(conn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===api_cftc_cit_supplement===\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 1/1 [00:00<00:00,  1.45it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===api_cftc_futures===\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 1/1 [00:02<00:00,  2.13s/it]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===api_cftc_combined===\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 1/1 [00:01<00:00,  1.55s/it]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===api_cftc_tff_futures===\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 1/1 [00:00<00:00,  2.19it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===api_cftc_tff_combined===\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 1/1 [00:00<00:00,  1.85it/s]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===api_cftc_disagg_futures===\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 17/17 [11:07<00:00, 39.24s/it]  \n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "===api_cftc_disagg_combined===\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "100%|██████████| 17/17 [11:24<00:00, 40.24s/it]  \n"
     ]
    }
   ],
   "source": [
    "\n",
    "tblnames = [\n",
    "    'api_cftc_cit_supplement',\n",
    "    'api_cftc_futures', 'api_cftc_combined',\n",
    "    'api_cftc_tff_futures', 'api_cftc_tff_combined',\n",
    "    'api_cftc_disagg_futures', 'api_cftc_disagg_combined'\n",
    "]\n",
    "for tblname in tblnames:\n",
    "    main(tblname, cur)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "9c5f8903d44efea2b789ca8382891403b65d0c111d781e48d4828ff2796f4d9c"
  },
  "kernelspec": {
   "display_name": "Python 3.8.12 64-bit (system)",
   "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.8.12"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}