修改创建 调整账目，置换币账目 接口

api/bill.py

@@ -30,14 +30,22 @@ async def create_pcf(
         fund_collect: AgnosticCollection = Depends(get_fund_collect),
         bill_collect: AgnosticCollection = Depends(get_bill_collect),
 ):
-    assets, adjust_assets, pending_assets, staking_assets = await query_fund_assets(fund_collect, item.fund_id, user.id)
-    if item.currency in assets:
     inc = item.volume if item.bill_type == PCFBillType.sub else -item.volume
+
+    assets, adjust_assets, pending_assets, staking_assets = await query_fund_assets(fund_collect, item.fund_id, user.id,
+                                                                                    FundStatus.active)
+    assets.setdefault(item.currency, 0)
+    # 如果是赎回 判断余额是否够
     assert assets[item.currency] + inc >= 0, "余额不足"
     assets[item.currency] += inc
-    else:
-        assert item.bill_type == PCFBillType.sub, "余额不足"
-        assets.update({item.currency: item.volume})
+
+    # if item.currency in assets:
+    #     inc = item.volume if item.bill_type == PCFBillType.sub else -item.volume
+    #     assert assets[item.currency] + inc >= 0, "余额不足"
+    #     assets[item.currency] += inc
+    # else:
+    #     assert item.bill_type == PCFBillType.sub, "余额不足"
+    #     assets.update({item.currency: item.volume})
     await update_assets(fund_collect, item.fund_id, assets=assets)
     pcf = PCFBill(user_id=user.id, **item.dict())
     await bill_collect.insert_one(pcf.dict())
@@ -50,34 +58,49 @@ async def create_pcf(
              summary='添加置换币账目',
              description='添加置换币账目')
 async def create_exchange(
-        item: CreateExchangeBill,
+        create_exchange_bill: CreateExchangeBill,
         user: User = Depends(get_current_user),
         fund_collect: AgnosticCollection = Depends(get_fund_collect),
         bill_collect: AgnosticCollection = Depends(get_bill_collect),
 ):
-    fund = await fund_collect.find_one({"id": item.fund_id, "user_id": user.id})
-    assert fund, NotFundError()
-    filter_asset = list(filter(lambda x: x["currency"] == item.input_currency, fund["assets"]))
-    assert filter_asset, f"{item.input_currency}余额不足"
-    input_asset = filter_asset[0]
-    assert input_asset["volume"] >= item.input_volume, f"{item.input_currency}余额不足"
-    update_input = UpdateOne(
-        {"id": item.fund_id, "assets.currency": item.input_currency},
-        {"$inc": {"assets.$.volume": -item.input_volume}}
-    )
-    output_filter = list(filter(lambda x: x["currency"] == item.output_currency, fund["assets"]))
-    update_output = UpdateOne(
-        {"id": item.fund_id, "assets.currency": item.output_currency},
-        {"$inc": {"assets.$.volume": item.output_volume}}
-    ) if output_filter else UpdateOne(
-        {"id": item.fund_id},
-        {"$push": {"assets": {"currency": item.output_currency, "volume": item.output_volume}}}
-    )
-    result = await fund_collect.bulk_write([update_input, update_output])
-    logger.info(result.modified_count)
-    input_value, output_value = item.input_volume * item.input_price, item.output_volume * item.output_price
+    assets, adjust_assets, pending_assets, staking_assets = await query_fund_assets(fund_collect,
+                                                                                    create_exchange_bill.fund_id,
+                                                                                    user.id,
+                                                                                    FundStatus.active)
+    assets.setdefault(create_exchange_bill.output_currency, 0)
+    assets.setdefault(create_exchange_bill.input_currency, 0)
+    assert assets[
+               create_exchange_bill.output_currency] >= create_exchange_bill.output_volume, f"{create_exchange_bill.output_currency}余额不足"
+
+    assets[create_exchange_bill.output_currency] -= create_exchange_bill.output_volume
+    assets[create_exchange_bill.input_currency] += create_exchange_bill.input_volume
+
+    await update_assets(fund_collect, create_exchange_bill.fund_id, assets=assets)
+
+    # fund = await fund_collect.find_one({"id": item.fund_id, "user_id": user.id})
+    # assert fund, NotFundError()
+    # filter_asset = list(filter(lambda x: x["currency"] == item.input_currency, fund["assets"]))
+    # assert filter_asset, f"{item.input_currency}余额不足"
+    # input_asset = filter_asset[0]
+    # assert input_asset["volume"] >= item.input_volume, f"{item.input_currency}余额不足"
+    # update_input = UpdateOne(
+    #     {"id": item.fund_id, "assets.currency": item.input_currency},
+    #     {"$inc": {"assets.$.volume": -item.input_volume}}
+    # )
+    # output_filter = list(filter(lambda x: x["currency"] == item.output_currency, fund["assets"]))
+    # update_output = UpdateOne(
+    #     {"id": item.fund_id, "assets.currency": item.output_currency},
+    #     {"$inc": {"assets.$.volume": item.output_volume}}
+    # ) if output_filter else UpdateOne(
+    #     {"id": item.fund_id},
+    #     {"$push": {"assets": {"currency": item.output_currency, "volume": item.output_volume}}}
+    # )
+    # result = await fund_collect.bulk_write([update_input, update_output])
+    # logger.info(result.modified_count)
+
+    input_value, output_value = create_exchange_bill.input_volume * create_exchange_bill.input_price, create_exchange_bill.output_volume * create_exchange_bill.output_price
     exchange_bill = ExchangeBill(user_id=user.id, input_value=input_value, output_value=output_value,
-                                 profit=output_value - input_value, **item.dict())
+                                 profit=output_value - input_value, **create_exchange_bill.dict())
     await bill_collect.insert_one(exchange_bill.dict())
     return Response[ExchangeBill](data=exchange_bill.dict())
 
@@ -88,23 +111,32 @@ async def create_exchange(
              summary='添加调整账目',
              description='添加调整账目')
 async def create_adjust(
-        item: CreateAdjustBill,
+        create_adjust_bill: CreateAdjustBill,
         user: User = Depends(get_current_user),
         fund_collect: AgnosticCollection = Depends(get_fund_collect),
         bill_collect: AgnosticCollection = Depends(get_bill_collect),
 ):
-    query = {"id": item.fund_id, "user_id": user.id}
-    result = await fund_collect.update_one(
-        {**query, "adjust_assets": {"$not": {"$elemMatch": {"currency": item.currency}}}},
-        {"$push": {"adjust_assets": {"currency": item.currency, "volume": item.volume}}}
-    )
-    if result.modified_count == 0:
-        inc_result = await fund_collect.update_one(
-            {**query, "adjust_assets.currency": item.currency},
-            {"$inc": {"adjust_assets.$.volume": item.volume}}
-        )
-        logger.info(f"inc_result={inc_result.modified_count}")
-    adjust_bill = AdjustBill(user_id=user.id, **item.dict())
+    assets, adjust_assets, pending_assets, staking_assets = await query_fund_assets(fund_collect,
+                                                                                    create_adjust_bill.fund_id, user.id,
+                                                                                    FundStatus.active)
+    adjust_assets.setdefault(create_adjust_bill.currency, 0)
+    adjust_assets.setdefault('fund_share', 0)
+    adjust_assets[create_adjust_bill.currency] += create_adjust_bill.volume
+    adjust_assets['fund_share'] += create_adjust_bill.fund_share
+
+    # query = {"id": item.fund_id, "user_id": user.id}
+    # result = await fund_collect.update_one(
+    #     {**query, "adjust_assets": {"$not": {"$elemMatch": {"currency": item.currency}}}},
+    #     {"$push": {"adjust_assets": {"currency": item.currency, "volume": item.volume}}}
+    # )
+    # if result.modified_count == 0:
+    #     inc_result = await fund_collect.update_one(
+    #         {**query, "adjust_assets.currency": item.currency},
+    #         {"$inc": {"adjust_assets.$.volume": item.volume}}
+    #     )
+    #     logger.info(f"inc_result={inc_result.modified_count}")
+    await update_assets(fund_collect, create_adjust_bill.fund_id, adjust_assets=adjust_assets)
+    adjust_bill = AdjustBill(user_id=user.id, **create_adjust_bill.dict())
     await bill_collect.insert_one(adjust_bill.dict())
     response = Response[AdjustBill](data=adjust_bill.dict())
     return response
@@ -121,18 +153,21 @@ async def create_staking_api(
         bill_collect: AgnosticCollection = Depends(get_bill_collect),
         fund_collect: AgnosticCollection = Depends(get_fund_collect)
 ):
-    staking_bill = await create_staking(create_staking_bill, user.id, bill_collect, fund_collect)
-
     assets, adjust_assets, pending_assets, staking_assets = await query_fund_assets(fund_collect,
                                                                                     user_id=user.id,
-                                                                                    fund_id=staking_bill.fund_id,
+                                                                                    fund_id=create_staking_bill.fund_id,
                                                                                     fund_status=FundStatus.active)
-    assert assets.get(create_staking_bill.currency, 0) >= create_staking_bill.volume, APIError(message='余额不足')
+    assert assets.get(create_staking_bill.currency, 0) >= create_staking_bill.volume, '余额不足'
     assets[create_staking_bill.currency] -= create_staking_bill.volume
     # 防止增加的币种没有 设置默认值
     pending_assets.setdefault(create_staking_bill.currency, 0)
     pending_assets[create_staking_bill.currency] += create_staking_bill.volume
     await update_assets(fund_collect, create_staking_bill.fund_id, pending_assets=pending_assets, assets=assets)
+
+    # 添加账目
+    staking_bill = StakingBill(user_id=user.id, **create_staking_bill.dict())
+    await bill_collect.insert_one(staking_bill.dict())
+
     response = Response[StakingBill](data=staking_bill.dict())
     return response
 

main.py

@@ -94,8 +94,8 @@ async def startup():
         misfire_grace_time=20
     )
 
-    if settings.env == 'LOCAL':
-        return
+    # if settings.env == 'LOCAL':
+    #     return
     app.state.scheduler.start()
 
     app.state.scheduler.print_jobs()

model/fund.py

@@ -6,6 +6,7 @@ from pydantic import Field
 from model import MyBaseModel
 from model.node import BaseNode
 from schema.fund import FundType, FundStatus
+from tools.time_helper import utc_now_timestamp
 
 
 class BaseFundItem(MyBaseModel):
@@ -29,6 +30,7 @@ class NormalFund(BaseFundItem):
     adjust_assets: Dict[str, float] = Field(default={}, description='调整账户持仓')
     pending_assets: Dict[str, float] = Field(default={}, description='pending资产')
     staking_assets: Dict[str, float] = Field(default={}, description='质押资产')
+    assets_update: float = Field(default_factory=utc_now_timestamp, description='余额更新时间')
 
 
 class StakingFund(BaseFundItem):
@@ -40,3 +42,4 @@ class StakingFund(BaseFundItem):
     adjust_assets: Dict[str, float] = Field(default={}, description='调整账户持仓')
     pending_assets: Dict[str, float] = Field(default={}, description='pending资产')
     staking_assets: Dict[str, float] = Field(default={}, description='质押资产')
+    assets_update: float = Field(default_factory=utc_now_timestamp, description='余额更新时间')

service/fund.py

@@ -2,6 +2,7 @@ from typing import Tuple
 
 from exception.db import NotFundError
 from schema.fund import FundStatus
+from tools.time_helper import utc_now_timestamp
 
 
 async def query_fund_assets(fund_collect, fund_id, user_id=None, fund_status=None) -> Tuple[dict, dict, dict, dict]:
@@ -17,10 +18,10 @@ async def query_fund_assets(fund_collect, fund_id, user_id=None, fund_status=Non
 
 
 # 修改资产
-async def update_assets(fund_collect, fund_id, assets=None, adjust_assets=None, pending_assets=None,
+async def update_assets(fund_collect, fund_id, *, assets=None, adjust_assets=None, pending_assets=None,
                         staking_assets=None):
     query = {'id': fund_id}
-    update_data = {}
+    update_data = {"assets_update": utc_now_timestamp()}
     if assets:
         update_data.update({"assets": assets})
     if adjust_assets:

service/scheduler.py

@@ -13,7 +13,7 @@ from dependencies import get_bill_collect, get_fund_collect
 from exception.http import RequestInvalidParamsError
 from schema.bill import StakingBillStatus, AllBillType
 from service.beacon import BeaconChaService
-from tools.time_helper import utc_now
+from tools.time_helper import utc_now_timestamp
 
 
 async def delete_task(job_id, scheduler):
@@ -67,7 +67,7 @@ async def update_staking_bill_status_task(beacon_service: BeaconChaService, mong
         if status:
             await bill_collect.find_one_and_update(
                 {'id': bill_item.id},
-                {'$set': {"status": status, "update_time": utc_now()}},
+                {'$set': {"status": status, "update_time": utc_now_timestamp()}},
                 return_document=ReturnDocument.AFTER)
             if status == StakingBillStatus.finish:
                 fund_collect = get_fund_collect(mongodb_manager)