agualpha 1.2.0

Python client for AGuAlpha Investment Platform API

AGuAlpha Python Client

Official Python library for accessing AGuAlpha Investment Platform data.

Installation

pip install agualpha

For async support:

pip install agualpha[async]

For pandas integration:

pip install agualpha[pandas]

Quick Start

Synchronous Usage

from agualpha import AGuAlphaClient

# Initialize client
client = AGuAlphaClient(api_key="sk-your-api-key-here")

# Get positions data
positions = client.get_positions()
print(f"Total positions: {positions.total}")

for position in positions.data:
    print(f"Date: {position.date}, Outstanding: {position.outstanding}")

# Get ideas data
ideas = client.get_ideas(status="active")
print(f"Total active ideas: {ideas.total}")

for idea in ideas.data:
    print(f"Symbol: {idea.ticker_symbol}, Status: {idea.status}")

# Close the connection
client.close()

Using Context Manager

from agualpha import AGuAlphaClient

with AGuAlphaClient(api_key="sk-your-api-key-here") as client:
    positions = client.get_positions(
        start_date="2024-01-01",
        end_date="2024-12-31"
    )
    print(f"Found {positions.total} positions")

Asynchronous Usage

import asyncio
from agualpha import AGuAlphaAsyncClient

async def main():
    async with AGuAlphaAsyncClient(api_key="sk-your-api-key-here") as client:
        # Fetch data concurrently
        positions, ideas = await asyncio.gather(
            client.get_positions(),
            client.get_ideas(status="active")
        )

        print(f"Positions: {positions.total}, Ideas: {ideas.total}")

asyncio.run(main())

Pandas Integration

from agualpha import AGuAlphaClient
from agualpha.utils import positions_to_dataframe, export_to_csv

client = AGuAlphaClient(api_key="sk-your-api-key-here")

# Get positions and convert to DataFrame
positions_response = client.get_positions()
df = positions_to_dataframe(positions_response.data)

# Export to CSV
export_to_csv(positions_response.data, "positions.csv")

CSI Weights

Read the full csi_weights table from the prices database. The data is returned as a pandas DataFrame for easy analysis.

from agualpha import AGuAlphaClient

with AGuAlphaClient(api_key="sk-your-api-key-here") as client:
    df = client.get_csi_weights_dataframe()
    print(df.head())
    print(fShape: {df.shape})

If you don't need pandas, you can get the raw list of dicts instead:

from agualpha import AGuAlphaClient

with AGuAlphaClient(api_key="sk-your-api-key-here") as client:
    rows = client.get_csi_weights()
    print(f"Total records: {len(rows)}")
    for row in rows:
        print(row)

API Reference

AGuAlphaClient

__init__(api_key: str, base_url: str = "http://47.98.139.211:8080/api")

Initialize the client with your API key.

get_positions(start_date: Optional[str] = None, end_date: Optional[str] = None) -> PositionsResponse

Get position data from subscribed analysts.

Parameters:

• start_date (str): Filter by start date (YYYY-MM-DD format)
• end_date (str): Filter by end date (YYYY-MM-DD format)

Returns: PositionsResponse

get_ideas(status: Optional[str] = None, direction: Optional[str] = None) -> IdeasResponse

Get trade ideas from subscribed analysts.

Parameters:

• status (str): Filter by status ("active", "closed")
• direction (str): Filter by direction ("long", "short")

Returns: IdeasResponse

get_csi_weights() -> list

Get all CSI weights data from the prices database (csi_weights table).

Returns: list of dicts, each representing a row from the csi_weights table.

get_csi_weights_dataframe() -> pandas.DataFrame

Convenience wrapper that returns the CSI weights data directly as a pandas DataFrame. Requires the pandas extra (pip install agualpha[pandas]).

Returns: pandas.DataFrame

Response Models

PositionsResponse

• success (bool): Request success status
• total (int): Total number of records
• data (List[Position]): List of position objects
• error (str, optional): Error message if failed

IdeasResponse

• success (bool): Request success status
• total (int): Total number of records
• data (List[Idea]): List of idea objects
• error (str, optional): Error message if failed

Error Handling

from agualpha import AGuAlphaClient
from agualpha.exceptions import APIError, AuthenticationError

try:
    client = AGuAlphaClient(api_key="invalid-key")
    positions = client.get_positions()
except AuthenticationError:
    print("Invalid API key")
except APIError as e:
    print(f"API error: {e}")

Requirements

• Python 3.8+
• requests 2.28.0+

License

MIT License