48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from pydantic import BaseModel
|
|
from pydantic.error_wrappers import ValidationError
|
|
|
|
|
|
class Args(BaseModel):
|
|
func_name: str = "ReturnOnNetAssets" # 指标函数名称
|
|
func_args: dict = {'net_profit': 1, 'current_period_total_owner_equity': 2, 'previous_period_total_owner_equity': 3} # 指标函数参数
|
|
|
|
|
|
app = FastAPI(
|
|
title="指标函数计算接口",
|
|
description="发送指标计算请求，返回计算后的指标数值",
|
|
version="v1.0.0"
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.post("/api/003/indicator_calculator/execute/", summary='输入参数，计算指标数值', description='接口描述', tags=['计算指标数值'])
|
|
async def root(args: Args):
|
|
|
|
try:
|
|
func_name = dict(args)["func_name"]
|
|
func_args = dict(args)["func_args"]
|
|
|
|
str1 = "from MyFunc.{} import {}".format(func_name, func_name)
|
|
str2 = "{}(**func_args).exec()".format(func_name)
|
|
|
|
exec(str1)
|
|
result = eval(str2)
|
|
|
|
return result
|
|
|
|
except ModuleNotFoundError:
|
|
raise HTTPException(status_code=400, detail="Function Not Found")
|
|
|
|
except ValidationError as e:
|
|
raise HTTPException(status_code=400, detail="FuncArgs Incorrect")
|