data_service_v2_1.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
import math
from decimal import Decimal
from app.api.engine import tamp_user_engine, tamp_product_engine, TAMP_SQL, tamp_fund_engine, tamp_diagnose_app_engine
from app.service.portfolio_diagnose import get_fund_substrategy


# from app.model.tamp_user_models import CustomerOrder, CustomerInfo
# from app.model.tamp_product_models import FundInfo
from app.utils.fund_rank import get_frequency, get_trade_cal


class UserCustomerDataAdaptor:
    user_id = ""
    customer_id = ""
    customer_real_name = ""
    month_date = ""
    end_date = ""
    group_data = {}
    trade_cal_date = None
    all_fund_id_list = None
    all_fund_type_dict = None
    all_fund_distribution = {}
    all_fund_performance = {}

    def __init__(self, user_id, customer_id, end_date=str(datetime.datetime.now()), index_id="IN0000007M"):
        self.user_id = user_id
        self.customer_id = customer_id
        self.compare_index_id = index_id
        self.valueSex = ''
        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.end_date = pd.to_datetime("2021-01-29")
        p_start_date = datetime.date(year=p_end_date.year, month=p_end_date.month, day=1)
        self.month_start_date = p_start_date
        # self.month_start_date = pd.to_datetime("2020-12-01")
        self.user_customer_order_df = self.get_user_customer_order_data()
        self.fund_nav_total, self.fund_cnav_total = self.get_customer_fund_nav_data()
        self.index_df = self.get_customer_index_nav_data()
        self.total_customer_order_cnav_df = self.total_combine_data()
        self.group_operate()

    @staticmethod
    def get_trade_cal(start_date, end_date):
        try:
            df = get_trade_cal()
            df = df[df["cal_date"] >= start_date]
            df.drop(['end_date'], 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
        except:
            pass

        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):

        with TAMP_SQL(tamp_fund_engine) as tamp_fund, TAMP_SQL(tamp_product_engine) as tamp_product, TAMP_SQL(tamp_diagnose_app_engine) as tamp_diagnose_app:
            tamp_diagnose_app_session = tamp_diagnose_app.session
            tamp_product_session = tamp_product.session
            tamp_fund_session = tamp_fund.session
            sql_user = """select f1.id, f1.fund_id, f2.realname,f3.customer_name,f3.valueSex,f1.type,f1.order_type,f1.pay_date,f1.subscription_fee,f1.confirm_share_date,f1.confirm_share,f1.confirm_amount,f1.nav,f1.folio_name, f1.comefrom from tamp_diagnose_app.customer_order_view f1, tamp_user.user_info f2,tamp_diagnose_app.customer_view f3   where f2.id=f1.user_id and f3.id=f1.customer_id  and f1.delete_tag=0 and user_id='{}' and customer_id='{}'""".format(self.user_id, self.customer_id)
            cur = tamp_diagnose_app_session.execute(sql_user)
            data = cur.fetchall()
            order_df = pd.DataFrame(list(data), columns=['order_id', 'fund_id', 'username', 'customer_name', 'valueSex', 'type', 'order_type', 'pay_date',
                                                         'subscription_fee', 'confirm_share_date', 'confirm_share',
                                                         'confirm_amount', 'nav', 'folio_name', 'comefrom'])
            cur_fund_id = list(order_df["fund_id"].unique())
            fund_list_str = str(cur_fund_id).replace("[", "(").replace("]", ")")

            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'])

            if len(cur_fund_id) > 0:
                sql_fund = "select distinct `id`, `fund_short_name`, `nav_frequency`, `substrategy` from `fund_info` where  `id` in {}".format(fund_list_str)
                cur = tamp_fund_session.execute(sql_fund)
                data = cur.fetchall()
                fund_df = pd.DataFrame(list(data), columns=['fund_id', 'fund_name', 'freq', 'substrategy'])

                if len(fund_df) > 0:
                    product_df = product_df.append(fund_df)
                    product_df = product_df.drop_duplicates("fund_id")
            fund_type3 = order_df[order_df["type"] == 3]
            if len(fund_type3) > 0:
                type3_fund_id = list(fund_type3["fund_id"].unique())
                fund_list_str_3 = str(type3_fund_id).replace("[", "(").replace("]", ")")
                sql_fund_3 = "select distinct `id`, `fund_name`, `substrategy` from `ifa_imported_fund_info` where  `id` in {}".format(
                    fund_list_str_3)
                cur = tamp_fund_session.execute(sql_fund_3)
                data = cur.fetchall()
                fund_df_3 = pd.DataFrame(list(data), columns=['fund_id', 'fund_name', 'substrategy'])
                fund_df_3["freq"] = 1

                if len(fund_df_3) > 0:
                    product_df = product_df.append(fund_df_3)
                    product_df = product_df.drop_duplicates("fund_id")

            fund_type0 = order_df[order_df["type"] == 0]
            if len(fund_type0) > 0:
                type0_fund_id = list(fund_type0["fund_id"].unique())
                fund_list_str_0 = str(type0_fund_id).replace("[", "(").replace("]", ")")
                sql_fund_0 = "select distinct `id`, `fund_short_name`, `substrategy` from `tx_fund_info` where  `id` in {}".format(
                    fund_list_str_0)
                cur = tamp_fund_session.execute(sql_fund_0)
                data = cur.fetchall()
                fund_df_0 = pd.DataFrame(list(data), columns=['fund_id', 'fund_name', 'substrategy'])
                fund_df_0["freq"] = 1

                if len(fund_df_0) > 0:
                    product_df = product_df.append(fund_df_0)
                    product_df = product_df.drop_duplicates("fund_id")
            user_customer_order_df = order_df.set_index('fund_id').join(product_df.set_index('fund_id')).reset_index()
            user_customer_order_df["substrategy"] = user_customer_order_df.apply(lambda x: get_fund_substrategy(x["fund_id"], x["type"]), axis=1)
            user_customer_order_df["confirm_share_date"] = user_customer_order_df["confirm_share_date"].apply(lambda x: pd.to_datetime(x.date()))
            self.customer_real_name = user_customer_order_df["customer_name"].values[0]
            self.ifa_real_name = user_customer_order_df["username"].values[0]
            self.valueSex = user_customer_order_df["valueSex"].values[0]
            user_customer_order_df = user_customer_order_df[user_customer_order_df["confirm_share_date"] <= self.end_date]
            user_customer_order_df["confirm_amount"] = user_customer_order_df["confirm_amount"] - user_customer_order_df["subscription_fee"]
            user_customer_order_df.sort_values(by="confirm_share_date", inplace=True)
            user_customer_order_df.index = pd.Series(range(len(user_customer_order_df)))
            user_customer_order_df["original_confirm_share"] = user_customer_order_df["confirm_share"]
            user_customer_order_df["divident_share"] = user_customer_order_df["confirm_share"]
            user_customer_order_df["reduce_share"] = user_customer_order_df["confirm_share"]
            user_customer_order_df["coefficient"] = Decimal(1.0)

            # 内部订单
            inter_df = user_customer_order_df[user_customer_order_df["comefrom"] == "inter"]
            inter_order_df = self.reduce_inter_order(inter_df)

            # 外部订单
            outside_df = user_customer_order_df[user_customer_order_df["comefrom"] == "app"]
            outside_order_df = self.reduce_outside_order(outside_df)
            # outside_order_df["comefrom"] = "app"
            # outside_order_df = outside_order_df[outside_order_df["fund_id"] == "HF00005AFK"]

            user_customer_order_df = pd.concat([inter_order_df, outside_order_df])

            user_customer_order_df = user_customer_order_df[((user_customer_order_df["order_type"] == 1)|(user_customer_order_df["order_type"] == 4))&(user_customer_order_df["confirm_share"] > 0)]
            user_customer_order_df.index = pd.Series(range(len(user_customer_order_df)))
            self.start_date = user_customer_order_df["confirm_share_date"].min()
            return user_customer_order_df

    # 核减内部订单
    def reduce_inter_order(self, order_df):
        inter_order_df = order_df.copy()
        inter_order_df.sort_values(by="confirm_share_date", inplace=True)

        def operate_type4(df):
            """
            处理同一天同一个基金的多笔分红记录（不同银行账号付款导致的作业流订单分红也会拆开成多笔）
            """
            df_row = []
            fund_dis_his = {}
            for index, row in df.iterrows():
                fund_id = row["fund_id"]
                if row["order_type"] != 4:
                    df_row.append(row)
                    continue
                dis_his = fund_dis_his.get(fund_id, None)
                if dis_his is None:
                    fund_dis_his[fund_id] = row
                    continue
                if row["confirm_share_date"] == dis_his["confirm_share_date"]:
                    dis_his["confirm_share"] += row["confirm_share"]
                    dis_his["original_confirm_share"] += row["original_confirm_share"]
                    dis_his["divident_share"] += row["divident_share"]
                    dis_his["reduce_share"] += row["reduce_share"]
            final_df = pd.DataFrame(data=df_row + list(fund_dis_his.values()))
            return final_df

        def reduce_order(fund_order, d_fund_id, d_date, d_nav):
            """
            再平衡分红后单子的份额数量，对分红前的单子进行拆分
            :param fund_order: 订单dataframe
            :param d_date: 分红订单日期
            :param d_nav: 分红后单位净值
            :return:
            """
            for _index, _row in fund_order.iterrows():
                if _row["confirm_share_date"] > d_date:
                    break
                if _row["order_type"] == 1 and _row["fund_id"] == d_fund_id and _row["nav"] > d_nav:
                    coefficient = _row["nav"] / d_nav
                    fund_order.loc[_index, "coefficient"] = coefficient * _row["coefficient"]
                    fund_order.loc[_index, "divident_share"] = _row["confirm_share"] * coefficient

            return fund_order

        if inter_order_df[inter_order_df["order_type"] == 4].groupby(["fund_id", "confirm_share_date"]).count()[
            "order_id"].max() > 1:
            inter_order_df = operate_type4(inter_order_df)
            inter_order_df.sort_values(by="confirm_share_date", inplace=True)

        for index, row in inter_order_df.iterrows():
            if row["order_type"] == 4:
                dividend_date = row["confirm_share_date"]
                dividend_nav = row["nav"]
                reduce_order(inter_order_df, row["fund_id"], dividend_date, dividend_nav)
                continue
            if row["order_type"] == 2:
                need_less_share = row["confirm_share"]
                for index_ori, row_ori in inter_order_df.iterrows():
                    if index_ori >= index:
                        break
                    if row["fund_id"] == row_ori["fund_id"] and (row_ori["order_type"] == 1 or row_ori["order_type"] == 4):
                        ori_share = row_ori["original_confirm_share"]
                        div_share = row_ori["divident_share"]
                        if need_less_share >= div_share:
                            need_less_share -= ori_share
                            inter_order_df.loc[index_ori, "divident_share"] = 0
                        else:
                            stay_share = row_ori["divident_share"] - need_less_share
                            inter_order_df.loc[index_ori, "divident_share"] = stay_share if stay_share > 0.01 else 0
                            need_less_share = 0
                        if need_less_share <= 0:
                            break
        inter_order_df["reduce_share"] = inter_order_df["divident_share"] / inter_order_df["coefficient"]
        inter_order_df["confirm_share"] = inter_order_df["reduce_share"]
        return inter_order_df

    # 核减外部订单
    def reduce_outside_order(self, order_df):
        outside_order_df = order_df.copy()
        for index, row in outside_order_df.iterrows():
            if row["order_type"] == 2:
                need_less_share = row["confirm_share"]
                for index_ori, row_ori in outside_order_df.iterrows():
                    if index_ori >= index:
                        break
                    if row["fund_id"] == row_ori["fund_id"] and row_ori["order_type"] == 1:
                        ori_share = row_ori["reduce_share"]
                        if need_less_share >= ori_share:
                            need_less_share -= ori_share
                            outside_order_df.loc[index_ori, "reduce_share"] = 0
                        else:
                            outside_order_df.loc[index_ori, "reduce_share"] = row_ori["reduce_share"] - need_less_share
                            need_less_share = 0
                        if need_less_share <= 0:
                            break
        outside_order_df["confirm_share"] = outside_order_df["reduce_share"]
        return outside_order_df

    # 获取客户持有的基金净值数据
    def get_customer_fund_nav_data(self):
        with TAMP_SQL(tamp_product_engine) as tamp_product, TAMP_SQL(tamp_fund_engine) as tamp_fund:
            tamp_product_session = tamp_product.session
            tamp_fund_session = tamp_fund.session
            now_date = datetime.datetime.now().strftime("%Y%m%d")
            trade_date_df = self.get_trade_cal("20000101", now_date)
            self.trade_cal_date = trade_date_df
            all_fund_nav = pd.DataFrame(index=trade_date_df["datetime"])
            all_fund_cnav = pd.DataFrame(index=trade_date_df["datetime"])

            temp_fund_id = []
            # for cur_fund_id in self.user_customer_order_df["fund_id"].unique():
            for index, row in self.user_customer_order_df.iterrows():
                cur_fund_id = row["fund_id"]
                if cur_fund_id in temp_fund_id:
                    continue
                else:
                    temp_fund_id.append(cur_fund_id)
                fund_type = row["type"]
                # 对应基金净值
                if fund_type == 0:
                    sql = """select distinct `price_date`, `nav`,`cumulative_nav` from `tx_fund_nav` where `fund_id`='{}' and `delete_tag`=0 order by `price_date` ASC""".format(cur_fund_id)
                    cur = tamp_fund_session.execute(sql)
                elif fund_type == 1:
                    sql = """select distinct `price_date`, `nav`,`cumulative_nav` from `fund_nav` where `fund_id`='{}' and `delete_tag`=0 order by `price_date` ASC""".format(
                        cur_fund_id)
                    cur = tamp_fund_session.execute(sql)
                elif fund_type == 2:
                    sql = """select distinct `price_date`, `nav`,`cumulative_nav` from `fund_nav` where `fund_id`='{}' and `delete_tag`=0 order by `price_date` ASC""".format(
                        cur_fund_id)
                    cur = tamp_product_session.execute(sql)
                elif fund_type == 3:
                    sql = """select distinct `price_date`, `nav`,`cumulative_nav` from `ifa_imported_fund_nav` where `fund_id`='{}'  and `delete_tag`=0 order by `price_date` ASC""".format(
                        cur_fund_id)
                    cur = tamp_fund_session.execute(sql)
                data = cur.fetchall()
                cur_fund_nav_df = pd.DataFrame(list(data), columns=['price_date', 'nav', 'cnav'])

                # # 对应基金分红
                dis_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)
                if fund_type == 0:
                    dis_sql = """select distinct `distribute_date`, `distribution` from `tx_fund_distribution` where `fund_id`='{}' and `distribute_type`='1' order by `distribute_date` ASC""".format(
                    cur_fund_id)
                    cur = tamp_fund_session.execute(dis_sql)
                else:
                    cur = tamp_product_session.execute(dis_sql)
                data = cur.fetchall()
                cur_fund_distribution_df = pd.DataFrame(list(data), columns=['price_date', 'distribution'])
                self.all_fund_distribution[cur_fund_id] = cur_fund_distribution_df

                # 对应基金performance数据
                per_sql = """select distinct `price_date`, `ret_1w`, `ret_cum_1m`, `ret_cum_6m`, `ret_cum_1y`, `ret_cum_ytd`, `ret_cum_incep` from `fund_performance` where `fund_id`='{}' order by `price_date` ASC""".format(
                    cur_fund_id)
                if fund_type == 0:
                    per_sql = """select distinct `price_date`, `ret_1w`, `ret_1m`, `ret_6m`, `ret_1y`, `ret_ytd`, `ret_incep` from `tx_fund_count` where `fund_id`='{}' order by `price_date` ASC""".format(
                        cur_fund_id)
                    cur = tamp_fund_session.execute(per_sql)
                elif fund_type == 3:
                    per_sql = """select distinct `price_date`, `ret_1w`, `ret_1m`, `ret_6m`, `ret_1y`, `ret_ytd`, `ret_incep` from `ifa_imported_fund_count` where `fund_id`='{}' order by `price_date` ASC""".format(
                        cur_fund_id)
                    cur = tamp_fund_session.execute(per_sql)
                else:
                    cur = tamp_product_session.execute(per_sql)
                data = cur.fetchall()
                cur_fund_performance_df = pd.DataFrame(list(data),
                columns=['price_date', 'ret_1w', 'ret_cum_1m', 'ret_cum_6m', 'ret_cum_1y', 'ret_cum_ytd', 'ret_cum_incep'])
                cur_fund_performance_df = cur_fund_performance_df[
                    cur_fund_performance_df["price_date"] <= self.end_date.date()]
                self.all_fund_performance[cur_fund_id] = cur_fund_performance_df

                cur_fund_nav_df["price_date"] = pd.to_datetime(cur_fund_nav_df["price_date"])
                cur_fund_nav_df.drop_duplicates(subset="price_date", keep='first', inplace=True)
                cur_fund_nav_df.set_index("price_date", inplace=True)
                cur_fund_nav_df = cur_fund_nav_df[cur_fund_nav_df.index.isin(all_fund_nav.index)]

                if math.isnan(row["freq"]):
                    try:
                        freq = get_frequency(cur_fund_nav_df)
                    except:
                        freq = 250
                    timeoffset_dict = {250: 1, 52: 5, 12: 30, 24: 15, 3: 120}
                    self.user_customer_order_df.loc[index, "freq"] = timeoffset_dict[freq]

                all_fund_nav[cur_fund_id] = cur_fund_nav_df["nav"]
                all_fund_cnav[cur_fund_id] = cur_fund_nav_df["cnav"]

            all_fund_nav = all_fund_nav[all_fund_nav.index <= self.end_date]
            all_fund_cnav = all_fund_cnav[all_fund_cnav.index <= self.end_date]
            # for cur_fund_id in self.user_customer_order_df["fund_id"].unique():
            #     all_fund_nav[cur_fund_id][all_fund_nav[cur_fund_id].apply(lambda x: math.isnan(x))]=np.nan
            #     all_fund_cnav[cur_fund_id][all_fund_cnav[cur_fund_id].apply(lambda x: math.isnan(x))] = np.nan
            self.last_nav_date = str(all_fund_cnav.dropna(how="all").index.values[-1])[:10]
            return all_fund_nav, all_fund_cnav

    # 获取客户对比指数净值数据
    def get_customer_index_nav_data(self, index_id="IN0000007M"):
        with TAMP_SQL(tamp_product_engine) as tamp_product:
            tamp_product_session = tamp_product.session
            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_cnav_total["index"] = index_df["index"]
            self.fund_nav_total["index"] = index_df["index"]
            self.index_df = index_df

            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())
            cur_folio_nav_df = self.fund_nav_total[fund_id_list]
            # fund_id_list.append("index")
            cur_folio_cnav_df = self.fund_cnav_total[fund_id_list]
            self.signal_folio_operate(folio, cur_folio_order_df, cur_folio_nav_df, cur_folio_cnav_df)
            continue

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

        p_order_df = p_order_df[(p_order_df["order_type"] == 1) | (p_order_df["order_type"] == 4)]
        p_nav_df = self.fund_nav_total.copy()
        p_cnav_df = self.fund_cnav_total.copy()

        start_date = pd.to_datetime(p_order_df["confirm_share_date"].min())
        nav_df = p_nav_df[p_nav_df.index >= start_date].copy()
        cnav_df = p_cnav_df[p_cnav_df.index >= start_date].copy()

        p_fund_id_list = list(p_order_df["fund_id"].unique())
        self.all_fund_id_list = p_fund_id_list
        self.all_fund_type_dict = {values["fund_id"]: values["type"] for values in
                                   p_order_df[["fund_id", "type"]].drop_duplicates().to_dict(orient='records')}
        for p_fund_id in p_fund_id_list:
            order_min_date = p_order_df[p_order_df["fund_id"] == p_fund_id]["confirm_share_date"].min()
            if pd.to_datetime(order_min_date) > start_date:
                cnav_df.loc[:order_min_date - datetime.timedelta(days=1), p_fund_id] = np.nan

        #  处理累积净值
        p_outside_order_df = p_order_df[p_order_df["comefrom"] == "app"]
        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 pd.isnull(p_nav_df.loc[confirm_share_date, cur_fund_id]):
                last_nav_data = p_nav_df[p_nav_df.index > confirm_share_date][cur_fund_id].dropna().head(1)
                last_cnav_data = p_cnav_df[p_cnav_df.index > confirm_share_date][cur_fund_id].dropna().head(1)
                # 判断上个净值日和当前确认日之中是否存在分红日
                """need add judge"""

                if len(last_nav_data) < 1:
                    cnav_df.loc[confirm_share_date, cur_fund_id] = row["nav"]
                else:
                    diff_nav = Decimal(row["nav"]) - Decimal(last_nav_data.values[0])
                    cur_cnav = last_cnav_data.values[0] + diff_nav
                    cnav_df.loc[confirm_share_date, cur_fund_id] = cur_cnav
            else:
                confirm_date_nav_data = p_nav_df[p_nav_df.index == confirm_share_date][cur_fund_id].tail(1)
                confirm_date_cnav_data = p_cnav_df[p_cnav_df.index == confirm_share_date][cur_fund_id].tail(1)
                diff_nav = Decimal(row["nav"]) - Decimal(confirm_date_nav_data.values[0])
                cur_cnav = Decimal(confirm_date_cnav_data.values[0]) + diff_nav
                cnav_df.loc[confirm_share_date, cur_fund_id] = cur_cnav

        cnav_df = cnav_df.dropna(axis=0, how="all").fillna(method='ffill')
        for index, row in p_outside_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 cnav_df:
                price = cnav_df[cur_fund_id].dropna()
                profit = price.diff().fillna(Decimal(0))
                cnav_df[cur_fund_id + "_profit"] = profit
                cnav_df[cur_fund_id + "_profit"] = cnav_df[cur_fund_id + "_profit"].fillna(Decimal(0))
                profit_ratio = profit / cnav_df[cur_fund_id].dropna().shift(1)
                cnav_df[cur_fund_id + "_profit_ratio"] = profit_ratio
                cnav_df[cur_fund_id + "_profit_ratio"] = cnav_df[cur_fund_id + "_profit_ratio"].fillna(Decimal(0))
                cnav_df[cur_fund_id + "_amount"] = 0
                cnav_df[cur_fund_id + "_earn"] = 0
                cnav_df[cur_fund_id + "_share"] = 0

            # buy
            if row['order_type'] == 1 or row['order_type'] == 4:
                # if row['order_type'] == 1:
                cnav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] += row["nav"] * row["confirm_share"]
                cnav_df.loc[confirm_share_date:, cur_fund_id + "_share"] += row["confirm_share"]

        #  处理单位净值
        nav_df = nav_df.dropna(axis=0, how="all").fillna(method='ffill')
        p_inter_order_df = p_order_df[p_order_df["comefrom"] == "inter"]
        for index, row in p_inter_order_df.iterrows():
            cur_fund_id = str(row["fund_id"])
            confirm_share_date = pd.to_datetime(row["confirm_share_date"])
            confirm_share_nav = row["nav"]

            nav_df.loc[confirm_share_date, cur_fund_id] = confirm_share_nav

        # 内部订单是用单位净值算收益
        nav_df = nav_df.fillna(method='ffill')
        actual_share_dict = {}
        for index, row in p_inter_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"] = Decimal(0)
                nav_df[cur_fund_id + "_profit_ratio"] = Decimal(0)
                nav_df[cur_fund_id + "_amount"] = 0
                nav_df[cur_fund_id + "_earn"] = 0
                nav_df[cur_fund_id + "_share"] = 0
                nav_df[cur_fund_id + "_reduce_share"] = 0

            # buy
            if row['order_type'] == 1:
                cur_fund_share = actual_share_dict.get(cur_fund_id, 0)
                cur_fund_share += row["original_confirm_share"] - (row["original_confirm_share"] * row["coefficient"] - row["divident_share"])
                nav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] += row["reduce_share"] * row["nav"]
                # nav_df.loc[confirm_share_date:, cur_fund_id + "_share"] = cur_fund_share
                nav_df.loc[confirm_share_date:, cur_fund_id + "_reduce_share"] += row["reduce_share"]
                actual_share_dict[cur_fund_id] = cur_fund_share

            if row['order_type'] == 4:
                cur_fund_share = actual_share_dict.get(cur_fund_id, 0)
                # cur_fund_share += row["reduce_share"]
                # nav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] = cur_fund_share * row["nav"]
                nav_df.loc[confirm_share_date:, cur_fund_id + "_share"] = cur_fund_share + row["reduce_share"]
                nav_df.loc[confirm_share_date:, cur_fund_id + "_reduce_share"] = cur_fund_share + row["reduce_share"]
                # actual_share_dict[cur_fund_id] = cur_fund_share

        for p_fund_id_ in p_outside_order_df["fund_id"].unique():
            cnav_df[p_fund_id_ + "_earn"] = (cnav_df[p_fund_id_ + "_profit"] * cnav_df[p_fund_id_ + "_share"]).apply(
                lambda x: float(x)).fillna(0)
            cnav_df[p_fund_id_ + "_cum_earn"] = cnav_df[p_fund_id_ + "_earn"].cumsum().fillna(0)
            cnav_df[p_fund_id_ + "_net_amount"] = cnav_df[p_fund_id_ + "_cum_earn"].apply(lambda x: Decimal(x)) + \
                                                  cnav_df[p_fund_id_ + "_amount"]

        # 内部订单
        for p_fund_id_ in p_inter_order_df["fund_id"].unique():
            nav_df[p_fund_id_ + "_net_amount"] = (nav_df[p_fund_id_] * nav_df[p_fund_id_ + "_reduce_share"]).apply(
                lambda x: float(x)).fillna(0)
            nav_df[p_fund_id_ + "_cum_earn"] = nav_df[p_fund_id_ + "_net_amount"] - nav_df[
                p_fund_id_ + "_amount"].apply(lambda x: float(x))
            nav_df[p_fund_id_ + "_earn"] = nav_df[p_fund_id_ + "_cum_earn"].diff().fillna(0)
            nav_df[p_fund_id_ + "_profit_ratio"] = (
                        nav_df[p_fund_id_ + "_earn"] / nav_df[p_fund_id_ + "_amount"].apply(lambda x: float(x))).apply(
                lambda x: Decimal(x))
            nav_df[p_fund_id_ + "_net_amount"] = nav_df[p_fund_id_ + "_net_amount"].apply(lambda x: Decimal(x))
            nav_df[p_fund_id_ + "_profit_ratio"] = nav_df[p_fund_id_ + "_profit_ratio"].apply(
                lambda x: Decimal(0) if math.isnan(x) else x)
            nav_df[p_fund_id_ + "_share"] = nav_df[p_fund_id_ + "_reduce_share"]

        finall_cnav_df = cnav_df.copy()
        all_nav_df = p_nav_df[p_nav_df.index >= start_date].copy().dropna(axis=0, how="all").fillna(method='ffill')
        for p_fund_id_ in p_order_df["fund_id"].unique():
            finall_cnav_df[p_fund_id_ + "_nav"] = all_nav_df[p_fund_id_]
            if p_fund_id_ + "_profit_ratio" not in finall_cnav_df.columns:
                finall_cnav_df[p_fund_id_ + "_profit_ratio"] = Decimal(0)
                finall_cnav_df[p_fund_id_ + "_amount"] = 0
                finall_cnav_df[p_fund_id_ + "_earn"] = 0
                finall_cnav_df[p_fund_id_ + "_share"] = 0
                finall_cnav_df[p_fund_id_ + "_net_amount"] = Decimal(0)
            # try:
            #     finall_cnav_df[p_fund_id_] = cnav_df[p_fund_id_]
            #     finall_cnav_df[p_fund_id_ + "_net_amount"] += cnav_df[p_fund_id_ + "_net_amount"]
            #     finall_cnav_df[p_fund_id_ + "_profit_ratio"] += cnav_df[p_fund_id_ + "_profit_ratio"]
            #     finall_cnav_df[p_fund_id_ + "_amount"] += cnav_df[p_fund_id_ + "_amount"]
            #     finall_cnav_df[p_fund_id_ + "_earn"] += cnav_df[p_fund_id_ + "_earn"]
            #     finall_cnav_df[p_fund_id_ + "_share"] += cnav_df[p_fund_id_ + "_share"]
            # except:
            #     pass
            try:
                finall_cnav_df[p_fund_id_ + "_nav"] = nav_df[p_fund_id_]
                finall_cnav_df[p_fund_id_ + "_net_amount"] += nav_df[p_fund_id_ + "_net_amount"]
                finall_cnav_df[p_fund_id_ + "_profit_ratio"] += nav_df[p_fund_id_ + "_profit_ratio"]
                finall_cnav_df[p_fund_id_ + "_amount"] += nav_df[p_fund_id_ + "_amount"]
                finall_cnav_df[p_fund_id_ + "_earn"] += nav_df[p_fund_id_ + "_earn"]
                finall_cnav_df[p_fund_id_ + "_share"] += nav_df[p_fund_id_ + "_share"]
            except:
                pass

        self.group_data[p_folio] = {"result_cnav_data": finall_cnav_df, "order_df": p_order_df}
        return cnav_df

    # 所有的数据操作
    def total_combine_data(self):

        p_order_df = self.user_customer_order_df[(self.user_customer_order_df["order_type"] == 1) |
                                                 (self.user_customer_order_df["order_type"] == 4)]
        p_nav_df = self.fund_nav_total.copy()
        p_cnav_df = self.fund_cnav_total.copy()

        start_date = pd.to_datetime(p_order_df["confirm_share_date"].min())
        nav_df = p_nav_df[p_nav_df.index >= start_date].copy()
        cnav_df = p_cnav_df[p_cnav_df.index >= start_date].copy()

        p_fund_id_list = list(p_order_df["fund_id"].unique())
        self.all_fund_id_list = p_fund_id_list
        self.all_fund_type_dict = {values["fund_id"]: values["type"] for values in p_order_df[["fund_id", "type"]].drop_duplicates().to_dict(orient='records')}
        for p_fund_id in p_fund_id_list:
            order_min_date = p_order_df[p_order_df["fund_id"] == p_fund_id]["confirm_share_date"].min()
            if pd.to_datetime(order_min_date) > start_date:
                cnav_df.loc[:order_min_date - datetime.timedelta(days=1), p_fund_id] = np.nan

        #  处理累积净值
        p_outside_order_df = p_order_df[p_order_df["comefrom"] == "app"]
        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 pd.isnull(p_nav_df.loc[confirm_share_date, cur_fund_id]):
                last_nav_data = p_nav_df[p_nav_df.index > confirm_share_date][cur_fund_id].dropna().head(1)
                last_cnav_data = p_cnav_df[p_cnav_df.index > confirm_share_date][cur_fund_id].dropna().head(1)
                # 判断上个净值日和当前确认日之中是否存在分红日
                """need add judge"""

                if len(last_nav_data) < 1:
                    cnav_df.loc[confirm_share_date, cur_fund_id] = row["nav"]
                else:
                    diff_nav = Decimal(row["nav"]) - Decimal(last_nav_data.values[0])
                    cur_cnav = last_cnav_data.values[0] + diff_nav
                    cnav_df.loc[confirm_share_date, cur_fund_id] = cur_cnav
            else:
                confirm_date_nav_data = p_nav_df[p_nav_df.index == confirm_share_date][cur_fund_id].tail(1)
                confirm_date_cnav_data = p_cnav_df[p_cnav_df.index == confirm_share_date][cur_fund_id].tail(1)
                diff_nav = Decimal(row["nav"]) - Decimal(confirm_date_nav_data.values[0])
                cur_cnav = Decimal(confirm_date_cnav_data.values[0]) + diff_nav
                cnav_df.loc[confirm_share_date, cur_fund_id] = cur_cnav

        cnav_df = cnav_df.dropna(axis=0, how="all").fillna(method='ffill')
        for index, row in p_outside_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 cnav_df:
                price = cnav_df[cur_fund_id].dropna()
                profit = price.diff().fillna(Decimal(0))
                cnav_df[cur_fund_id + "_profit"] = profit
                cnav_df[cur_fund_id + "_profit"] = cnav_df[cur_fund_id + "_profit"].fillna(Decimal(0))
                profit_ratio = profit / cnav_df[cur_fund_id].dropna().shift(1)
                cnav_df[cur_fund_id + "_profit_ratio"] = profit_ratio
                cnav_df[cur_fund_id + "_profit_ratio"] = cnav_df[cur_fund_id + "_profit_ratio"].fillna(Decimal(0))
                cnav_df[cur_fund_id + "_amount"] = 0
                cnav_df[cur_fund_id + "_earn"] = 0
                cnav_df[cur_fund_id + "_share"] = 0

            # buy
            if row['order_type'] == 1 or row['order_type'] == 4:
                # if row['order_type'] == 1:
                cnav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] += row["nav"] * row["confirm_share"]
                cnav_df.loc[confirm_share_date:, cur_fund_id + "_share"] += row["confirm_share"]

        #  处理单位净值
        nav_df = nav_df.dropna(axis=0, how="all").fillna(method='ffill')
        p_inter_order_df = p_order_df[p_order_df["comefrom"] == "inter"]
        for index, row in p_inter_order_df.iterrows():
            cur_fund_id = str(row["fund_id"])
            confirm_share_date = pd.to_datetime(row["confirm_share_date"])
            confirm_share_nav = row["nav"]

            nav_df.loc[confirm_share_date, cur_fund_id] = confirm_share_nav

        # 内部订单是用单位净值算收益
        nav_df = nav_df.fillna(method='ffill')
        actual_share_dict = {}
        for index, row in p_inter_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"] = Decimal(0)
                nav_df[cur_fund_id + "_profit_ratio"] = Decimal(0)
                nav_df[cur_fund_id + "_amount"] = 0
                nav_df[cur_fund_id + "_earn"] = 0
                nav_df[cur_fund_id + "_share"] = 0
                nav_df[cur_fund_id + "_reduce_share"] = 0

            # buy
            if row['order_type'] == 1:
                cur_fund_share = actual_share_dict.get(cur_fund_id, 0)
                cur_fund_share += row["original_confirm_share"] - (row["original_confirm_share"] * row["coefficient"] - row["divident_share"])
                nav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] += row["reduce_share"] * row["nav"]
                nav_df.loc[confirm_share_date:, cur_fund_id + "_share"] = cur_fund_share
                nav_df.loc[confirm_share_date:, cur_fund_id + "_reduce_share"] += row["reduce_share"]
                actual_share_dict[cur_fund_id] = cur_fund_share

            if row['order_type'] == 4:
                cur_fund_share = actual_share_dict.get(cur_fund_id, 0)
                # cur_fund_share += row["reduce_share"]
                # nav_df.loc[confirm_share_date:, cur_fund_id + "_amount"] = cur_fund_share * row["nav"]
                nav_df.loc[confirm_share_date:, cur_fund_id + "_share"] = cur_fund_share + row["reduce_share"]
                nav_df.loc[confirm_share_date:, cur_fund_id + "_reduce_share"] = cur_fund_share + row["reduce_share"]
                # actual_share_dict[cur_fund_id] = cur_fund_share

        for p_fund_id_ in p_outside_order_df["fund_id"].unique():
            cnav_df[p_fund_id_ + "_earn"] = (cnav_df[p_fund_id_ + "_profit"] * cnav_df[p_fund_id_ + "_share"]).apply(lambda x: float(x)).fillna(0)
            cnav_df[p_fund_id_ + "_cum_earn"] = cnav_df[p_fund_id_ + "_earn"].cumsum().fillna(0)
            cnav_df[p_fund_id_ + "_net_amount"] = cnav_df[p_fund_id_ + "_cum_earn"].apply(lambda x: Decimal(x)) + cnav_df[p_fund_id_ + "_amount"]
        for p_fund_id_ in p_inter_order_df["fund_id"].unique():
            nav_df[p_fund_id_ + "_net_amount"] = (nav_df[p_fund_id_] * nav_df[p_fund_id_ + "_reduce_share"]).apply(lambda x: float(x)).fillna(0)
            nav_df[p_fund_id_ + "_cum_earn"] = nav_df[p_fund_id_ + "_net_amount"] - nav_df[p_fund_id_ + "_amount"].apply(lambda x: float(x))
            nav_df[p_fund_id_ + "_earn"] = nav_df[p_fund_id_ + "_cum_earn"].diff().fillna(0)
            nav_df[p_fund_id_ + "_profit_ratio"] = (nav_df[p_fund_id_ + "_earn"] / nav_df[p_fund_id_ + "_amount"].apply(lambda x: float(x))).apply(lambda x: Decimal(x))
            nav_df[p_fund_id_ + "_net_amount"] = nav_df[p_fund_id_ + "_net_amount"].apply(lambda x: Decimal(x))
            nav_df[p_fund_id_ + "_profit_ratio"] = nav_df[p_fund_id_ + "_profit_ratio"].apply(lambda x: Decimal(0) if math.isnan(x) else x)

        finall_cnav_df = cnav_df.copy()
        all_nav_df = p_nav_df[p_nav_df.index >= start_date].copy().dropna(axis=0, how="all").fillna(method='ffill')
        for p_fund_id_ in p_order_df["fund_id"].unique():
            finall_cnav_df[p_fund_id_ + "_nav"] = all_nav_df[p_fund_id_]
            if p_fund_id_ + "_profit_ratio" not in finall_cnav_df.columns:
                finall_cnav_df[p_fund_id_ + "_profit_ratio"] = Decimal(0)
                finall_cnav_df[p_fund_id_ + "_amount"] = 0
                finall_cnav_df[p_fund_id_ + "_earn"] = 0
                finall_cnav_df[p_fund_id_ + "_share"] = 0
                finall_cnav_df[p_fund_id_ + "_net_amount"] = Decimal(0)
            # try:
            #     finall_cnav_df[p_fund_id_] = cnav_df[p_fund_id_]
            #     finall_cnav_df[p_fund_id_ + "_net_amount"] += cnav_df[p_fund_id_ + "_net_amount"]
            #     finall_cnav_df[p_fund_id_ + "_profit_ratio"] += cnav_df[p_fund_id_ + "_profit_ratio"]
            #     finall_cnav_df[p_fund_id_ + "_amount"] += cnav_df[p_fund_id_ + "_amount"]
            #     finall_cnav_df[p_fund_id_ + "_earn"] += cnav_df[p_fund_id_ + "_earn"]
            #     finall_cnav_df[p_fund_id_ + "_share"] += cnav_df[p_fund_id_ + "_share"]
            # except:
            #     pass
            try:
                finall_cnav_df[p_fund_id_ + "_nav"] = nav_df[p_fund_id_]
                finall_cnav_df[p_fund_id_ + "_net_amount"] += nav_df[p_fund_id_ + "_net_amount"]
                finall_cnav_df[p_fund_id_ + "_profit_ratio"] += nav_df[p_fund_id_ + "_profit_ratio"]
                finall_cnav_df[p_fund_id_ + "_amount"] += nav_df[p_fund_id_ + "_amount"]
                finall_cnav_df[p_fund_id_ + "_earn"] += nav_df[p_fund_id_ + "_earn"]
                finall_cnav_df[p_fund_id_ + "_share"] += nav_df[p_fund_id_ + "_share"]
            except:
                pass

        return finall_cnav_df