73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
|
# ruff: noqa: E402
|
||
|
|
|
||
|
|
"""Print a customer's read-only dynamic ETF allocation result for local verification."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
from dataclasses import asdict
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
sys.path.insert(0, str(ROOT))
|
||
|
|
|
||
|
|
from app.core.contracts import RequestContext
|
||
|
|
from app.service.asset_allocation_service import AssetAllocationService
|
||
|
|
from app.service.customer_profile_service import CustomerProfileService
|
||
|
|
|
||
|
|
|
||
|
|
def parse_args() -> argparse.Namespace:
|
||
|
|
parser = argparse.ArgumentParser(description="Verify dynamic allocation for one local customer")
|
||
|
|
parser.add_argument("customer_id", help="Existing customer ID with a completed profile and goal")
|
||
|
|
parser.add_argument(
|
||
|
|
"--market-metrics",
|
||
|
|
action="store_true",
|
||
|
|
help="Also print the classified market metrics used by the optimizer",
|
||
|
|
)
|
||
|
|
return parser.parse_args()
|
||
|
|
|
||
|
|
|
||
|
|
async def verify(customer_id: str) -> dict[str, object]:
|
||
|
|
context = RequestContext(
|
||
|
|
user_id=customer_id,
|
||
|
|
trace_id="local-dynamic-allocation-verification",
|
||
|
|
roles=("customer",),
|
||
|
|
permissions=(
|
||
|
|
"asset-allocation:generate:self",
|
||
|
|
"customer-profile:read:self",
|
||
|
|
"investment-goal:read:self",
|
||
|
|
),
|
||
|
|
)
|
||
|
|
return await AssetAllocationService().generate_for_agent(context)
|
||
|
|
|
||
|
|
|
||
|
|
async def market_metrics(customer_id: str) -> list[dict[str, object]]:
|
||
|
|
context = RequestContext(
|
||
|
|
user_id=customer_id,
|
||
|
|
trace_id="local-dynamic-allocation-metrics",
|
||
|
|
roles=("customer",),
|
||
|
|
permissions=("customer-profile:read:self",),
|
||
|
|
)
|
||
|
|
profile = await CustomerProfileService().current_for_agent(context)
|
||
|
|
if profile is None or not isinstance(profile.get("risk_level"), str):
|
||
|
|
return []
|
||
|
|
return [
|
||
|
|
asdict(item)
|
||
|
|
for item in await AssetAllocationService()._market_metrics(profile["risk_level"])
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
args = parse_args()
|
||
|
|
result = asyncio.run(verify(args.customer_id))
|
||
|
|
if args.market_metrics:
|
||
|
|
result["market_metrics"] = asyncio.run(market_metrics(args.customer_id))
|
||
|
|
print(json.dumps(result, ensure_ascii=False, default=str, indent=2))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|