ind/Ind_Model_Base.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting Ind_Model_Base.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile Ind_Model_Base.py\n",
    "\n",
    "import sys\n",
    "import os \n",
    "module_path = os.path.abspath(os.path.join('..')) \n",
    "if module_path not in sys.path: \n",
    "    sys.path.append(module_path)\n",
    "    \n",
    "import pandas as pd\n",
    "import copy\n",
    "\n",
    "import QUANTAXIS as QA\n",
    "\n",
    "\n",
    "from base.Constants import LOW_FREQUENCE\n",
    "from QAIndicatorStructExt import QA_DataStruct_Indicators_Ext\n",
    "\n",
    "class Ind_Model:\n",
    "    # （静态）最优描述符号，指标人工训练结束后的记录，用于给后续更大的模型引用时做提示\n",
    "#     optimum_param={'valid':False, 'main':'main', 'desition_direct':1, 'window':14, 'freq':'d','neutralize':{'enable':False,'static_mv':False}}\n",
    "\n",
    "    def __init__(self,data_df, ind_name, frequence=QA.FREQUENCE.DAY, pramas_default=None):\n",
    "        \"\"\"\n",
    "        注意：默认是 fast_mode 的，为了批量生成因子的时候加速。\n",
    "        \"\"\"\n",
    "#         if not isinstance(data, type(QA.OUTPUT_FORMAT.DATASTRUCT)):\n",
    "#             raise TypeError('Must be DATASTRUCT')\n",
    "        if pramas_default is None:\n",
    "            self._pramas_default = self.on_set_params_default()\n",
    "        else:\n",
    "            self._pramas_default = pramas_default\n",
    "            \n",
    "        if not isinstance(self._pramas_default, dict):\n",
    "            raise TypeError('_pramas_default MUST BE DICT')\n",
    "        \n",
    "        self.fast_mode = True\n",
    "        self.pramas = copy.deepcopy(self._pramas_default)\n",
    "        self.ind_name = ind_name\n",
    "        self.data = data_df\n",
    "        self.ind_df = None\n",
    "        self.desition_df = None\n",
    "        self.frequence = frequence\n",
    "        \n",
    "\n",
    "    def __repr__(self):\n",
    "        return '< {} in pramas  {} ,{},{} >'.format(self.ind_name, \n",
    "                                                    str (self.pramas), \n",
    "                                                    'not fit' if not self.is_fitted else 'fitted:'+ str(self.keys_ind), \n",
    "                                                    'not desition' if not self.has_desition else 'has desition')\n",
    "        \n",
    "    def change_pramas(self,**dic):\n",
    "        for k in dic.keys():\n",
    "            self.pramas[k] = dic[k]\n",
    "    \n",
    "    def reset_pramas(self):\n",
    "        self.pramas = copy.deepcopy(self._pramas_default)\n",
    "    \n",
    "    @property\n",
    "    def cur_pramas(self):\n",
    "        return self.pramas\n",
    "    @property\n",
    "    def default_pramas(self):\n",
    "        return self._pramas_default\n",
    "    @property\n",
    "    def keys_pramas(self):\n",
    "        return self._pramas_default.keys()\n",
    "    @property\n",
    "    def keys_ind(self):\n",
    "        if not self.is_fitted:\n",
    "            raise Exception(\"need fit the model first\")\n",
    "        return self.ind_df.keys().to_list()\n",
    "    \n",
    "    @property\n",
    "    def is_fitted(self):\n",
    "        return not self.ind_df is None\n",
    "    \n",
    "    @property\n",
    "    def has_desition(self):\n",
    "        return not self.desition_df is None\n",
    "    \n",
    "    @property\n",
    "    def is_low_frequence(self):\n",
    "        return self.frequence in LOW_FREQUENCE\n",
    "    \n",
    "\n",
    "    @property\n",
    "    def ind(self):\n",
    "        if not self.is_fitted:\n",
    "            raise Exception(\"need fit the model first\")\n",
    "        return QA_DataStruct_Indicators_Ext(self.ind_df)\n",
    "\n",
    "    @property\n",
    "    def desition(self):\n",
    "        if not self.has_desition  or not isinstance(self.desition_df, pd.DataFrame):\n",
    "            print(\"desition_df:\",self.desition_df)\n",
    "            raise Exception('on_desition_structure error')\n",
    "        return QA_DataStruct_Indicators_Ext(self.desition_df)\n",
    "    \n",
    "    def set_fast_mode(self, is_fast = True):\n",
    "        self.fast_mode = is_fast\n",
    "    \n",
    "\n",
    "    def on_set_params_default(self) -> dict:\n",
    "        #初始化时设置默认参数\n",
    "        raise NotImplementedError\n",
    "        \n",
    "    def on_indicator_structuring(self, data) -> pd.DataFrame:\n",
    "        #构造因子\n",
    "        raise NotImplementedError\n",
    "        \n",
    "    def on_desition_structuring(self, data, ind_data) -> pd.DataFrame:\n",
    "        #生成因子结论，非必要\n",
    "        raise NotImplementedError\n",
    "        \n",
    "        \n",
    "        \n",
    "    def excute_for_multicode(self, data, func, **pramas):\n",
    "        return data.groupby(level=1, as_index=False, group_keys=False).apply(func,**pramas)\n",
    "\n",
    "    def fit(self):\n",
    "        self.ind_df = self.on_indicator_structuring(self.data)\n",
    "        \n",
    "        try:\n",
    "            self.desition_df = self.on_desition_structuring(self.data, self.ind_df)\n",
    "        except NotImplementedError:\n",
    "            self.desition_df = None\n",
    "        except Exception as err:\n",
    "            raise Exception(err)\n",
    "            \n",
    "    def checking(self,benchmark=None):\n",
    "        print('evaluation table')\n",
    "        "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}