蛋卷基金交易记录导出

README.md

使用方法

1. 登录蛋卷基金网页版：https://danjuanapp.com/
2. 打开浏览器 DevTools 的 Network 面板，刷新页面后，在 Network 面板点开任意一个返回结果为 json 类型的请求，复制 请求Cookie 中的内容到 config.json 中

3. 执行 python danjuan_exporter.py -c config.json -o orders.json

danjuan_exporter.py

import json
from copy import deepcopy
from datetime import datetime
from operator import itemgetter
from urllib.parse import urljoin
from collections import defaultdict

import click
import requests


class DanjuanExporter:

    BASE_URL = 'https://danjuanapp.com'
    ORDER_LIST_PATH = '/djapi/order/p/list'
    FUND_TRADE_DETAIL_PATH = '/djapi/fund/order/{order_id}'
    PLAN_TRADE_DETAIL_PATH = '/djapi/order/p/plan/{order_id}'
    PLAN_SUBORDER_DETAIL_PATH = '/djapi/plan/order/{order_id}'
    ACTION_MAPPING = {
        '买入': 'buy',
        '定投': 'buy',
        '分红': 'reinvest',
        '卖出': 'sell',
        '组合转出': 'sell',
        '组合转入': 'buy',
    }

    def __init__(self, config):
        self.cookies = config
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:90.0) Gecko/20100101 Firefox/90.0'
        }
        self.http = requests.Session()

    def _http_get(self, path, params=None):
        url = urljoin(self.BASE_URL, path)
        resp = self.http.get(url, params=params, headers=self.headers, cookies=self.cookies)
        resp.raise_for_status()
        return resp.json()

    def get_trade_detail(self, trade_type, order_id):
        if trade_type == 'fund':
            path = self.FUND_TRADE_DETAIL_PATH.format(order_id=order_id)
        elif trade_type == 'plan':
            path = self.PLAN_TRADE_DETAIL_PATH.format(order_id=order_id)
        elif trade_type == 'plan-suborder':
            path = self.PLAN_SUBORDER_DETAIL_PATH.format(order_id=order_id)
        else:
            return None

        detail = self._http_get(path)['data']
        return detail

    def parse_fund_order(self, data):
        result = {
            'action': self.ACTION_MAPPING[data['action_desc']],
            'time': datetime.fromtimestamp(int(data['created_at'] / 1000)),
            'code': data['fd_code'] + '.OF',
            'name': data['fd_name'],
            'amount': data['confirm_volume'],
        }
        confirm_info = dict([
            item.split(',') for item in data['confirm_infos'][0][1:]
        ])
        result['price'] = float(confirm_info['确认净值'])
        if confirm_info['手续费'] != '--':
            result['fee'] = float(confirm_info['手续费'].replace('元', ''))
        else:
            result['fee'] = 0.00

        if confirm_info['确认金额'] != '--':
            result['money'] = float(confirm_info['确认金额'].replace('元', ''))
        else:
            result['money'] = round(result['amount'] * result['price'] + result['fee'], 2)

        return result

    def parse_plan_order(self, data):
        orders = defaultdict(list)
        for sub_order in data['sub_order_list']:
            for order in sub_order['orders']:
                action = self.ACTION_MAPPING[order['action_desc']]
                detail = self.get_trade_detail('plan-suborder', order['order_id'])
                if len(detail['confirm_infos']) == 1:
                    confirm_info = dict([
                        item.split(',') for item in detail['confirm_infos'][0][1:]
                    ])
                    order_info = {
                        'plan': {'name': order['plan_name'], 'code': order['plan_code']},
                        'action': action,
                        'time': datetime.fromtimestamp(int(detail['created_at'] / 1000)),
                        'code': detail['fd_code'] + '.OF',
                        'name': detail['fd_name'],
                        'amount': float(confirm_info['确认份额'].replace('份', '')),
                        'money': float(confirm_info['确认金额'].replace('元', '')),
                        'price': float(confirm_info['确认净值']),
                        'fee': float(confirm_info['手续费'].replace('元', '')),
                    }
                    orders[
                        (order_info['code'], order_info['time'], order_info['action'])
                    ].append(order_info)
                    continue

                funds = [
                    {'code': order['fd_code'] + '.OF', 'name': order['fd_name']},
                    {'code': order['target_fd_code'] + '.OF', 'name': order['target_fd_name']},
                ]
                sell_fund, buy_fund = funds if action == 'sell' else funds[::-1]
                for confirm_info in detail['confirm_infos']:
                    info = dict([item.split(',') for item in confirm_info[1:]])
                    if confirm_info[0].startswith('转出'):
                        cur_action, cur_fund = 'sell', sell_fund
                    else:
                        cur_action, cur_fund = 'buy', buy_fund

                    order_info = {
                        'plan': {'name': order['plan_name'], 'code': order['plan_code']},
                        'action': cur_action,
                        'time': datetime.fromtimestamp(int(detail['created_at'] / 1000)),
                        'amount': float(info['确认份额'].replace('份', '')),
                        'money': float(info['确认金额'].replace('元', '')),
                        'price': float(info['确认净值']),
                        'fee': float(info['手续费'].replace('元', '')),
                        **cur_fund
                    }
                    orders[
                        (order_info['code'], order_info['time'], order_info['action'])
                    ].append(order_info)

        result = []
        for sub_orders in orders.values():
            merged_order = deepcopy(sub_orders[0])
            merged_order['amount'] = sum([order['amount'] for order in sub_orders])
            merged_order['money'] = sum([order['money'] for order in sub_orders])
            merged_order['fee'] = sum([order['fee'] for order in sub_orders])
            result.append(merged_order)

        result.sort(key=itemgetter('time'))
        return result

    def list_orders(self):
        resp = self._http_get(self.ORDER_LIST_PATH, {'page': 1, 'type': 'all'})
        if resp['result_code'] != 0:
            return []

        result = []
        for item in resp['data']['items']:
            if item['status'] != 'success':
                continue

            detail = self.get_trade_detail(item['ttype'], item['order_id'])
            if item['ttype'] == 'fund':
                result.append(self.parse_fund_order(detail))
            elif item['ttype'] == 'plan':
                result.extend(self.parse_plan_order(detail))

        result.sort(key=itemgetter('time'))
        return result


@click.command()
@click.option("-c", "--config-file", required=True)
@click.option("-o", "--outfile")
def main(config_file, outfile):
    config = json.load(open(config_file))
    exporter = DanjuanExporter(config)
    orders = exporter.list_orders()
    if outfile:
        with open(outfile, 'w') as f:
            json.dump(orders, f, ensure_ascii=False, indent=2)
    else:
        for order in orders:
            print(order)


if __name__ == '__main__':
    main()