增加重新计算净值接口，修改净值记录接口

api/nav.py

 import datetime
-
-from fastapi import APIRouter, Depends
+from typing import Optional
+from fastapi import APIRouter, Depends, Query, Body
+from fastapi.background import BackgroundTasks
 from motor.core import AgnosticCollection
-
-from dependencies import get_nav_collect
+from dependencies import get_nav_collect, get_fund_collect
 from model import SortParams, FilterTime, Page, PageResponse, Response
 from model.fund import StakingFundNav
-from service.nav import calculate_nav
+from service.nav import calculate_nav, build_fund_nav
 
 router = APIRouter()
 
 
 @router.get("/{fund_id}/", response_model=PageResponse[StakingFundNav], summary="查询净值记录")
-async def nav(
+async def get_nav(
         fund_id: str,
         sort_by: SortParams = Depends(SortParams),
         filter_time: FilterTime = Depends(FilterTime),
@@ -34,8 +34,58 @@ async def nav(
 @router.post("/{fund_id}/", response_model=Response[StakingFundNav], summary="插入净值记录")
 async def insert_nav(
         fund_id: str,
-        calc_time: datetime.datetime,
+        calc_time: Optional[datetime.datetime] = Query(default=None, title="时间"),
 ):
-    fund_nav = await calculate_nav(fund_id=fund_id, calc_time=calc_time)
+    fund_nav = await calculate_nav(fund_id=fund_id, calc_time=calc_time) if calc_time else await calculate_nav(fund_id)
     response = Response[StakingFundNav](data=fund_nav)
     return response
+
+
+@router.post("/", response_model=Response[bool], summary="新增净值记录")
+async def create_nav(
+        fund_id: str = Body(..., title="基金id"),
+        record_date: datetime.datetime = Body(..., title="记录日期"),
+        nav: float = Body(..., title="净值"),
+        fund_share: float = Body(..., title="份额"),
+        fund_collect: AgnosticCollection = Depends(get_fund_collect),
+        nav_collect: AgnosticCollection = Depends(get_nav_collect),
+):
+    fund_data = await fund_collect.find({"id": fund_id})
+    fund_nav = build_fund_nav(fund_data, record_date, fund_share, nav)
+    insert = await nav_collect.insert_one(fund_nav)
+    response = Response(data=insert.acknowledged)
+    return response
+
+
+@router.put("/{nav_id}/", response_model=Response[int], summary="修改净值记录")
+async def update_nav(
+        nav_id: str,
+        fund_id: str = Body(..., title="基金id"),
+        record_date: datetime.datetime = Body(..., title="记录日期"),
+        nav: float = Body(..., title="净值"),
+        fund_share: float = Body(..., title="份额"),
+        fund_collect: AgnosticCollection = Depends(get_fund_collect),
+        nav_collect: AgnosticCollection = Depends(get_nav_collect),
+):
+    update_result = await nav_collect.update_one(
+        {"id": nav_id, "fund_id": fund_id},
+        {"$set": {"record_date": record_date, "nav": nav, "fund_share": fund_share}}
+    )
+    response = Response(data=update_result.modified_count)
+    return response
+
+
+@router.put("/recalculate/{fund_id}/", response_model=Response[bool], summary="重新计算净值")
+async def recalculate_nav(
+        fund_id: str,
+        background_task: BackgroundTasks,
+        start: datetime.datetime = Query(..., title="开始时间"),
+        end: datetime.datetime = Query(None, title="结束时间"),
+):
+    end = end or datetime.datetime.utcnow()
+    delta = end - start
+    for i in range(delta.days + 1):
+        date = start + datetime.timedelta(days=i)
+        background_task.add_task(calculate_nav, fund_id=fund_id, calc_time=date, update_fund=False)
+    response = Response(data=True)
+    return response

service/nav.py

 import datetime
+import time
 import uuid
 
 import pytz
 from loguru import logger
+from pymongo import ReturnDocument
 
 from dependencies import get_fund_collect, get_cmc_price_redis, get_hour_price_collect, get_bill_collect, \
     get_nav_collect
@@ -11,17 +13,36 @@ from service.beacon import BeaconChaService
 from service.price import get_price
 
 
-async def calculate_nav(fund_id, calc_time: datetime.datetime = None, beach_service=BeaconChaService()):
+def build_fund_nav(fund_data: dict, record_date, total_fund_share, net_value):
+    fund_nav = {}
+    for key, value in fund_data.items():
+        if key == "_id":
+            continue
+        elif key == "id":
+            fund_nav["fund_id"] = value
+            fund_nav["id"] = uuid.uuid1().__str__()
+        else:
+            fund_nav[key] = value
+    fund_nav.update(
+        {"record_date": record_date, "fund_share": total_fund_share, "nav": net_value, "update_time": int(time.time())})
+    return fund_nav
+
+
+async def calculate_nav(fund_id, calc_time: datetime.datetime = None, beach_service=BeaconChaService(),
+                        update_fund=True):
     # todo 测试
-    calc_time = calc_time.replace(minute=0, second=0, microsecond=0,
-                                  tzinfo=pytz.UTC) if calc_time else datetime.datetime.utcnow().replace(minute=0,
-                                                                                                        second=0,
-                                                                                                        microsecond=0,
-                                                                                                        tzinfo=pytz.UTC)
     from main import app
     fund_collect = get_fund_collect(app.state.mongodb_manager)
+    nav_collect = get_nav_collect(app.state.mongodb_manager)
     logger.info(f'[定时任务开始执行] [计算净值] {fund_id} ')
-    fund_data = await fund_collect.find_one({'id': fund_id, 'fund_status': FundStatus.active.value})
+    # 判断当天是否有净值记录，如果有则根据
+    fund_record = await fund_collect.find_one({'id': fund_id, 'fund_status': FundStatus.active.value})
+    record_date = (calc_time or datetime.datetime.utcnow()).replace(hour=0, minute=0, second=0, microsecond=0,
+                                                                    tzinfo=pytz.UTC)
+    if fund_record["settlement_time"] == "00:00" or fund_record["settlement_time"] == "0:00":
+        record_date -= datetime.timedelta(days=1)
+    nav_record = await nav_collect.find_one({"fund_id": fund_id, "record_date": record_date})
+    fund_data = nav_record or fund_record
     amount = {}
     # 查询asset资产
     for key, value in fund_data['assets'].items():
@@ -29,7 +50,7 @@ async def calculate_nav(fund_id, calc_time: datetime.datetime = None, beach_serv
         amount[key] += value
 
     # 查询在途资产
-    for key, value in fund_data['assets'].items():
+    for key, value in fund_data['pending_assets'].items():
         amount.setdefault(key, 0)
         amount[key] += value
 
@@ -72,36 +93,17 @@ async def calculate_nav(fund_id, calc_time: datetime.datetime = None, beach_serv
         -item["fund_share"] if item["bill_type"] == "redemption" else item["fund_share"] for item in result)
     value = sum(nav.values())
     net_value = round(value / total_fund_share, 2) if total_fund_share > 0 else fund_data["base_nav"]
-    await fund_collect.update_one({'id': fund_id}, {"$set": {"nva": net_value}})
-    nav_collect = get_nav_collect(app.state.mongodb_manager)
-
-    date = calc_time.replace(hour=0, minute=0, second=0, microsecond=0,
-                             tzinfo=pytz.UTC) if calc_time else datetime.datetime.utcnow().replace(hour=0, minute=0,
-                                                                                                   second=0,
-                                                                                                   microsecond=0,
-                                                                                                   tzinfo=pytz.UTC)
-    if fund_data["settlement_time"] == "00:00":
-        date -= datetime.timedelta(days=1)
-    nav_record = await nav_collect.find_one({"fund_id": fund_id, "record_date": date})
+    if update_fund:
+        await fund_collect.update_one({'id': fund_id}, {"$set": {"nav": net_value}})
     if nav_record:
-        error = f'重复净值记录 [{nav_record}] [mongo] [{date}]'
-        logger.error(error)
-        raise Exception(error)
-    fund_nav = {}
-    for key, value in fund_data.items():
-        if key == "_id":
-            continue
-        elif key == "id":
-            fund_nav["fund_id"] = value
-            fund_nav["id"] = uuid.uuid1().__str__()
-        else:
-            fund_nav[key] = value
-    fund_nav["record_date"] = date
-    fund_nav["fund_share"] = total_fund_share
-    fund_nav["nav"] = net_value
-    insert = await nav_collect.insert_one(fund_nav)
-    if not insert.acknowledged:
-        error = f'写入净值记录失败 [{fund_nav}] [mongo] [{date}]'
-        logger.error(error)
-        raise Exception(error)
-    return fund_nav
+        fund_nav = nav_record.copy()
+        fund_nav.update({"record_date": record_date, "fund_share": total_fund_share, "nav": net_value, "update_time": int(time.time())})
+    else:
+        fund_nav = build_fund_nav(fund_record, record_date, total_fund_share, net_value)
+    update_record = await nav_collect.find_one_and_update(
+        {"fund_id": fund_id, "record_date": record_date},
+        {"$set": fund_nav},
+        upsert=True,
+        return_document=ReturnDocument.AFTER
+    )
+    return update_record