ARTICLE · 1048903
[创新技术阁][AiTrader]使用定时任务定时更新账户资产
[创新技术阁][AiTrader]使用定时任务定时更新账户资产
相关内容： [创新技术阁][AiTrader]新增交易所管理并修改前端显示 [创新技术阁][AiTrader]新增交易对Symbol的管理 [创新技术阁][AiTrader]完善Symbol显示 [创新技术阁][AiTrader]新增Apikey的管理 [创新技术阁][AiTrader]完善Apikey显示和验证 [创新技术阁][AiTrader]新增账户管理以及问题修复 [创新技术阁][AiTrader]Apikey新增获取option接口并完善账户显示
1. 新增账户信息
在新增账户信息的时候，我们需要考虑如下几点：
判断今日是否已有记录，如果有则更新，没有则新增。 如果是第一条记录，那么利润和总利润都是0； profit (当日利润) = 今日总资产 - 上一条记录总资产 (若无记录则为0)。 total_profit (累计利润) = 上一条记录累计利润 + 今日利润。具体实现：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 async def create(self, data: QuantAccountCreateSchema) -> QuantAccountOutSchema: """ 创建或更新账户信息 逻辑: 1. 判断今日是否已有记录，如果有则更新，没有则新增。 2. 如果是第一条记录，那么利润和总利润都是0； 3. profit (当日利润) = 今日总资产 - 上一条记录总资产 (若无记录则为0)。 4. total_profit (累计利润) = 上一条记录累计利润 + 今日利润。 """ if isinstance(data.update_date, date): today_date = data.update_date elif isinstance(data.update_date, str): try: today_date = date.fromisoformat(data.update_date.split()[0].split("T")[0]) except Exception: today_date = date.today() else: today_date = date.today() crud = QuantAccountCRUD(self.auth, self.db) # 查找今日之前最近的一条历史记录 stmt_last = ( select(QuantAccountModel) .where( QuantAccountModel.exchange_id == data.exchange_id, QuantAccountModel.apikey_id == data.apikey_id, QuantAccountModel.update_date < today_date, QuantAccountModel.is_deleted == False, ) .order_by(QuantAccountModel.update_date.desc(), QuantAccountModel.id.desc()) .limit(1) ) last_record = (await self.db.execute(stmt_last)).scalars().first() today_total = float(data.total_balance_usdt) if last_record is None: # 2. 如果是第一条记录，那么利润和总利润都是0 profit = 0.0 total_profit = 0.0 else: # 3. profit (当日利润) = 今日总资产 - 上一条记录总资产 (若无记录则为0) profit = round(today_total - float(last_record.total_balance_usdt), 2) # 4. total_profit (累计利润) = 上一条记录累计利润 + 今日利润 total_profit = round(float(last_record.total_profit or 0.0) + profit, 2) # 1. 判断今日是否已有记录，如果有则更新，没有则新增 today_obj = await crud.get( exchange_id=data.exchange_id, apikey_id=data.apikey_id, update_date=today_date, ) if today_obj: update_data = data.model_dump(exclude_unset=True, exclude={"id"}) update_data["profit"] = profit update_data["total_profit"] = total_profit update_data["update_date"] = today_date obj = await crud.update(id=today_obj.id, data=update_data) else: data.profit = profit data.total_profit = total_profit data.update_date = today_date obj = await crud.create(data=data) return QuantAccountOutSchema.model_validate(obj) 2.定时任务功能函数
定时更新账户的资产，我们需要先从数据库读取所有需要更新账户的apikey，然后循环根据apikey从交易所获取资产进行保存。
在app/plugin/module_quant/下新增一个job_func的包，然后新增一个文件sync_account.py来实现获取账户资产的功能：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 #!/usr/bin/env python# encoding: utf-8# @version: v1.0# @author: Kandy.Ye# @contact: Kandy.Ye@outlook.com# @file: sync_account.py# @time: 2026/09/20 11:21from datetime import datetime, date, timedelta, timezoneimport ccxtfrom app.core.logger import loggerfrom app.core.database import async_db_sessionfrom app.core.base_schema import AuthSchemafrom app.plugin.module_quant.apikey.service import QuantApikeyServicefrom app.plugin.module_quant.exchange.service import QuantExchangeServicefrom app.plugin.module_quant.account.service import QuantAccountServicefrom app.plugin.module_quant.account.schema import QuantAccountCreateSchemaasync def sync_account(): return await SyncAccount().main()class SyncAccount(object): def __init__(self): self.exchanges = {} self.apikeys = [] async def init(self, auth: AuthSchema, db): exchanges = await QuantExchangeService(auth=auth, db=db).get_list(search={}) for exchange in exchanges: self.exchanges[exchange.id] = exchange apikeys = await QuantApikeyService(auth=auth, db=db).get_list(search={}) for apikey in apikeys: self.apikeys.append(apikey) async def get_total_balance(self, client: ccxt.Exchange, total: dict): total_balance = 0 for symbol, amount in total.items(): if amount <= 0.00001: continue if symbol == 'USDT': total_balance += amount else: try: ticker = client.fetch_ticker(f"{symbol}/USDT") price = ticker.get('last', 0) total_balance += amount * price except Exception: pass return total_balance async def main(self): async with async_db_session() as session: auth = AuthSchema(db=session) await self.init(auth=auth, db=session) feedback = '' for apikey in self.apikeys: exchange = self.exchanges[apikey.exchange_id] exchange_class = getattr(ccxt, exchange.name) exchange_options = { 'apiKey': apikey.api_key, 'secret': apikey.secret_key, 'enableRateLimit': True, 'verbose': False, } if exchange.name in ('okx', 'bitget'): exchange_options['password'] = apikey.passphrase try: client = exchange_class(exchange_options) balance = client.fetch_balance() logger.info(balance) info = balance.get('info') if exchange.name.lower() == 'binance': details = {key: value for key, value in balance.items() if type(value) is dict and 'total' in value and value['total'] > 0} param = { 'total_balance_usdt': float(balance.get('USDT', {}).get('total', 0)), 'total': {key: value for key, value in balance.get('total', {}).items() if value > 0}, 'free': {key: value for key, value in balance.get('free', {}).items() if value > 0}, 'used': {key: value for key, value in balance.get('used', {}).items() if value > 0}, 'details': [], 'upl': "0", 'update_date': date.today().strftime("%Y-%m-%d"), 'profit': 0, 'total_profit': 0, 'spot_balance_usdt': float(balance.get('USDT', {}).get('total', 0)), 'future_balance_usdt': 0, } for key, value in details.items(): param['details'].append({ "eq": float(value.get('total')), "ccy": key, "availEq": float(value.get('free')), "frozenBal": float(value.get('used')), "eqUsd": '0' }) elif exchange.name == 'bitget': if type(info) == list and len(info) == 0: continue param = { 'total_balance_usdt': float(info[0].get('frozen')) + float(info[0].get('available')), 'total': balance.get('total'), 'free': balance.get('free'), 'used': balance.get('used'), 'details': [], 'upl': "0", 'update_date': date.today().strftime("%Y-%m-%d"), 'profit': 0, 'total_profit': 0, 'spot_balance_usdt': float(info[0].get('frozen')) + float(info[0].get('available')), 'future_balance_usdt': 0, } for item in info: param['details'].append({ "eq": float(item.get('frozen')) + float(item.get('available')), "ccy": item.get('coin'), "availEq": float(item.get('available')), "frozenBal": float(item.get('frozen')), "eqUsd": '0' }) else: param = { 'total_balance_usdt': float(info.get('data')[0].get('totalEq')), 'total': balance.get('total'), 'free': balance.get('free'), 'used': balance.get('used'), 'details': info.get('data')[0].get('details'), 'upl': "0", 'update_date': (datetime.strptime(balance.get('datetime'), "%Y-%m-%dT%H:%M:%S.%fZ").date() + timedelta(hours=8)).strftime("%Y-%m-%d"), 'profit': 0, 'total_profit': 0, 'spot_balance_usdt': float(info.get('data')[0].get('totalEq')), 'future_balance_usdt': 0, } param['exchange_id'] = apikey.exchange_id param['apikey_id'] = apikey.id param['total_balance_usdt'] = await self.get_total_balance(client, param.get('total')) param['created_id'] = apikey.created_id param['updated_id'] = apikey.updated_id param['status'] = '0' now_utc8_str = datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d %H:%M:%S') param['created_time'] = now_utc8_str param['updated_time'] = now_utc8_str await QuantAccountService(auth=auth, db=session).create(data=QuantAccountCreateSchema(**param)) await session.commit() feedback = feedback + f'获取{apikey.name}余额成功 ' except Exception as e: await session.rollback() logger.error(f"Error fetching balance for {apikey.name}: {e}") feedback = feedback + f'获取{apikey.name}余额失败 ' logger.info(feedback) return feedback 这里需要注意，由于不同的交易所返回的信息不一样，所以我们需要根据交易所进行特定字段的解析，而且有些交易所并没有给出总资产的USDT价值，也需要我们根据返回的资产进行转换。
3.测试获取账户资产
在tests下新增一个文件test_sync_account.py来测试我们写的获取账户资产的功能，内容如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 #!/usr/bin/env python # encoding: utf-8 # @version: v1.0 # @author: Kandy.Ye # @contact: Kandy.Ye@outlook.com # @file: sync_account.py # @time: 2026/09/20 11:21 import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from app.core.base_model import MappedBase from app.utils.import_util import ImportUtil # 加载全部 ORM 模型以解析关系映射（如 UserMixin 引用的 UserModel） ImportUtil.find_models(MappedBase) from app.plugin.module_quant.job_fun.sync_account import sync_account from app.core.database import async_engine async def test_sync_account(): try: result = await sync_account() print(result) finally: await async_engine.dispose() if __name__ == "__main__": import asyncio asyncio.run(test_sync_account()) 运行效果如下：
4. 新建定时任务节点
启动前后端服务，在菜单任务管理->定时任务->节点管理里面新建一个节点：
节点名称：同步账户资产 节点编码：sync_account 执行计划：因为我们需要每天早上8:10更新，所以Cron表达式 Cron表达式：0 10 8 * * ?，可以通过弹出的对话框进行选择，自动生成表达式，如下图 
开始时间：弹出的对话框里面选择今天 结束时间：可以不填 存储器：默认Redis 执行器：线程池 合并运行：是，错过执行时间后使用运行一次 最大实例：1 处理器：注意必须要定义handler函数。
1 2 3 4 5 6 7 8 9 def handler(**kwargs): import asyncio from app.core.ap_scheduler import scheduler from app.plugin.module_quant.job_fun.sync_account import sync_account loop = getattr(scheduler, '_eventloop', None) if loop and loop.is_running(): return asyncio.run_coroutine_threadsafe(sync_account(), loop).result() return asyncio.run(sync_account())