async def getUserFills(
address: Optional[str] = None,
start_time: Optional[datetime.datetime] = None,
end_time: Optional[datetime.datetime] = None,
limit: Optional[int] = None, # Add this for `limit`
offset: Optional[int] = None, # Add this for `offset`
timeout_ms: Optional[int] = None,
) -> Optional[Dict[str, Any]]:
base_url = os.getenv("API_SERVER_URL")
if not base_url:
raise ValueError("API_SERVER_URL environment variable not set")
# Default address filter to authenticated account when available.
if address is None:
address = os.getenv("ACCOUNT")
async with httpx.AsyncClient() as session:
client = PerpetualsClient(base_url=base_url, session=session)
try:
start_time_pb = None
end_time_pb = None
if start_time is not None:
start_time_pb = Timestamp()
start_time_pb.FromDatetime(start_time)
if end_time is not None:
end_time_pb = Timestamp()
end_time_pb.FromDatetime(end_time)
fills = await client.get_user_fills(
address=address,
start_time=start_time_pb,
end_time=end_time_pb,
limit=limit,
offset=offset,
timeout_ms=timeout_ms or client.timeout_ms,
)
if fills is not None:
fills_list = [
MessageToDict(fill, always_print_fields_with_no_presence=True)
for fill in fills
]
fills_list.sort(key=lambda f: f.get("timestamp", 0), reverse=True)
return {"fills": fills_list}
return None
except Exception as e:
print(f"Error getting user fills asdf: {e}")
print("\nTroubleshooting:")
print(" - Is the server running?")
print(" - Is a valid market_id provided?")
print(" - Check network connectivity")
return None