data_service.py

#!/usr/bin/python3.6
# -*- coding: utf-8 -*-
# @Time    : 2020/11/18 19:12
# @Author  : Jie. Z
# @Email   : zhaojiestudy@163.com
# @File    : data_service.py
# @Software: PyCharm

import pandas as pd
import numpy as np
from sqlalchemy import and_
import tushare as ts
import datetime
from app.api.engine import tamp_user_session, tamp_product_session
# from app.model.tamp_user_models import t_customer_order, t_customer_info
# from app.model.tamp_product_models import t_fund_info


class UserCustomerDataAdaptor:
    user_id = ""
    customer_id = ""
    start_date = ""
    end_date = ""
    group_data = {}

    def __init__(self, user_id, customer_id, end_date=str(datetime.date.today())):
        self.user_id = user_id
        self.customer_id = customer_id
        p_end_date = pd.to_datetime(end_date).date()
        p_end_date = datetime.date(year=p_end_date.year, month=p_end_date.month, day=1) - datetime.timedelta(days=1)
        self.end_date = pd.to_datetime(str(p_end_date))
        self.user_customer_order_df = self.get_user_customer_order_data()
        self.fund_nav_total = self.get_customer_fund_nav_data()
        self.get_customer_index_nav_data()
        self.group_operate()

    @staticmethod
    def get_trade_cal(start_date, end_date):
        ts.set_token('ac1f734f8a25651aa07319ca35b1b0c0854e361e306fe85d85e092bc')
        pro = ts.pro_api()
        if end_date is not None:
            df = pro.trade_cal(exchange='SSE', start_date=start_date, end_date=end_date, is_open='1')
        else:
            df = pro.trade_cal(exchange='SSE', start_date=start_date, is_open='1')
        df.drop(['exchange', 'is_open'], axis=1, inplace=True)
        df.rename(columns={'cal_date': 'end_date'}, inplace=True)
        df["datetime"] = df["end_date"].apply(lambda x: datetime.datetime.strptime(x, "%Y%m%d"))
        return df

    # 获取理财师下该用户所有订单列表
    def get_user_customer_order_data(self):
        # data1 = tamp_user_session.query(t_customer_order)\
        #     .filter(user_id = self.user_id).all()
        # data2 = tamp_user_session.query(t_customer_info).all()
        # data3 = tamp_product_session.query(t_fund_info).all()

        sql_user = """select f2.realname,f3.customer_name,fund_id,f1.order_type,f1.pay_date,f1.subscription_fee,f1.confirm_share_date,f1.confirm_share,f1.confirm_amount,f1.nav,f1.folio_name from customer_order f1, user_info f2,customer_info f3   where f2.id=f1.user_id and f3.id=f1.customer_id and f1.user_id='{}' and f1.customer_id='{}'""".format(self.user_id, self.customer_id)
        cur = tamp_user_session.execute(sql_user)
        data = cur.fetchall()
        order_df = pd.DataFrame(list(data), columns=['username', 'customer_name', 'fund_id', 'order_type', 'pay_date',
                                                     'subscription_fee', 'confirm_share_date', 'confirm_share',
                                                     'confirm_amount', 'nav', 'folio_name'])

        sql_product = "select distinct `id`, `fund_short_name`, `nav_frequency`, `substrategy` from `fund_info`"
        cur = tamp_product_session.execute(sql_product)
        data = cur.fetchall()
        product_df = pd.DataFrame(list(data), columns=['fund_id', 'fund_name', 'freq', 'substrategy'])

        user_customer_order_df = order_df.set_index('fund_id').join(product_df.set_index('fund_id')).reset_index()
        self.start_date = user_customer_order_df["confirm_share_date"].min()
        return user_customer_order_df

    # 获取客户持有的基金净值数据
    def get_customer_fund_nav_data(self):
        now_date = datetime.datetime.now().strftime("%Y%m%d")
        trade_date_df = self.get_trade_cal("20000101", now_date)
        all_fund_nav = pd.DataFrame(index=trade_date_df["datetime"])

        for cur_fund_id in self.user_customer_order_df["fund_id"].unique():
            # 对应基金净值
            sql = """select distinct `price_date`, `nav`,`cumulative_nav` from `fund_nav` where `fund_id`='{}'  order by `price_date` ASC""".format(cur_fund_id)
            cur = tamp_product_session.execute(sql)
            data = cur.fetchall()
            cur_fund_nav_df = pd.DataFrame(list(data), columns=['price_date', 'nav', 'cnav'])

            # # 对应基金分红
            # sql = """select distinct `distribute_date`, `distribution` from `fund_distribution` where `fund_id`='{}' and `distribute_type`='1' order by `distribute_date` ASC""".format(
            #     cur_fund_id)
            # cur = tamp_product_session.execute(sql)
            # data = cur.fetchall()
            # cur_fund_distribution_df = pd.DataFrame(list(data), columns=['price_date', 'distribution'])

            cur_fund_nav_df["price_date"] = pd.to_datetime(cur_fund_nav_df["price_date"])

            cur_fund_nav_df.set_index("price_date", inplace=True)
            all_fund_nav[cur_fund_id] = cur_fund_nav_df["cnav"]
        all_fund_nav = all_fund_nav[all_fund_nav.index <= self.end_date]
        return all_fund_nav

    # 获取客户对比指数净值数据
    def get_customer_index_nav_data(self, index_id="IN0000007M"):

        sql = "select distinct price_date,close from fund_market_indexes where index_id='{}'  order by price_date ASC".format(index_id)
        cur = tamp_product_session.execute(sql)
        data = cur.fetchall()
        index_df = pd.DataFrame(list(data), columns=['price_date', 'index'])
        index_df["price_date"] = pd.to_datetime(index_df["price_date"])
        index_df.set_index("price_date", inplace=True)
        self.fund_nav_total["index"] = index_df["index"]

        return index_df

    # 分组合计算
    def group_operate(self):

        for folio in self.user_customer_order_df["folio_name"].unique():
            cur_folio_order_df = self.user_customer_order_df[self.user_customer_order_df["folio_name"] == folio]
            fund_id_list = list(self.user_customer_order_df["fund_id"].unique())
            fund_id_list.append("index")
            cur_folio_nav_df = self.fund_nav_total[fund_id_list]
            self.signal_folio_operate(folio, cur_folio_order_df, cur_folio_nav_df)
            continue

    # 单个组合数据操作
    def signal_folio_operate(self, p_folio, p_order_df, p_nav_df):
        start_date = pd.to_datetime(p_order_df["confirm_share_date"].min())
        nav_df = p_nav_df[p_nav_df.index >= start_date].copy()

        for index, row in p_order_df.iterrows():
            cur_fund_id = str(row["fund_id"])
            confirm_share_date = pd.to_datetime(row["confirm_share_date"])

            if cur_fund_id+"_amount" not in nav_df:
                nav_df[cur_fund_id + "_profit"] = (nav_df[cur_fund_id].dropna() - nav_df[cur_fund_id].dropna().shift(1))
                nav_df[cur_fund_id+"_amount"] = 0
                nav_df[cur_fund_id + "_earn"] = 0
                nav_df[cur_fund_id + "_share"] = 0

            # buy
            if row['order_type'] == 1:
                nav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] += row["confirm_amount"]
                nav_df.loc[confirm_share_date:, cur_fund_id + "_share"] += row["confirm_share"]
            # sell
            elif row['order_type'] == 2:
                nav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] -= row["confirm_amount"]
                nav_df.loc[confirm_share_date:, cur_fund_id + "_share"] -= row["confirm_share"]

            nav_df[cur_fund_id + "_earn"] = nav_df[cur_fund_id + "_profit"] * nav_df[cur_fund_id + "_share"]
        self.group_data[p_folio] = {"basic_data": nav_df}
        return nav_df