TickFlow Python SDK

All symbol-based queries (quotes, K-lines, etc.) use a unified format: Code.MarketSuffix (using a dot as a separator).

Supported Market Suffixes

Examples

• A-shares: 600000.SH, 000001.SZ, 920662.BJ
• ETFs: 510300.SH, 159915.SZ
• Indices: 000001.SH, 399006.SZ
• US Stocks: AAPL.US
• HK Stocks: 00700.HK

You can install the TickFlow SDK using pip.

To include full support for pandas DataFrames and progress bars, install the [all] extra. Otherwise, install the base version for a minimal footprint.

Requirements:

• Python 3.9+
• Python 3.10+ is recommended

# Install with DataFrame and progress bar support
pip install "tickflow[all]" --upgrade

# Install basic version
pip install tickflow

Both SyncAPIClient and AsyncAPIClient implement automatic retries to handle transient failures.

Retryable Exceptions

The client will automatically retry requests if the following errors occur:

• ConnectionError (Network connection issues)
• TimeoutError (Request timeouts)
• InternalServerError (HTTP 5xx status codes)
• RateLimitError (HTTP 429 status codes)

Backoff Strategy

The client uses exponential backoff with jitter to prevent overwhelming the server:

1. Exponential Backoff: The delay increases as base_delay * (2**attempt) (e.g., 1s, 2s, 4s, 8s...).
2. Jitter: A random variation of $\pm 25%$ is added to the delay to prevent synchronized retry spikes.
3. Cap: The delay is capped at a max_delay (defaulting to 30 seconds).

The TickFlow SDK uses a special sentinel value NOT_GIVEN to differentiate between an argument that was explicitly passed as None and an argument that was omitted entirely. This is critical when calling API methods where None has a specific semantic meaning (e.g., clearing a filter) versus not providing the parameter at all (using the server's default).

To check if a value was explicitly provided, use the is_given() function. To clean a dictionary of these sentinels before processing, use strip_not_given().

from tickflow import NOT_GIVEN, is_given, strip_not_given

# Example: Checking if a value is provided
val = NOT_GIVEN
print(is_given(val))  # False

# Example: Cleaning a parameter dictionary
params = {"symbol": "AAPL", "limit": NOT_GIVEN}
clean_params = strip_not_given(params)
# clean_params is now {"symbol": "AAPL"}

The stream attribute provides a unified WebSocket streaming interface (/v1/ws/stream) that supports quotes and depth channels.

Note: The realtime attribute is a legacy, deprecated quote-only stream (/v1/ws/quotes). Use stream instead.

from tickflow import TickFlow

client = TickFlow(api_key="your-api-key")
stream = client.stream

@stream.on_quotes
def handle(quotes):
    for q in quotes:
        print(f"{q['symbol']}: {q['last_price']}")

stream.subscribe("quotes", ["600000.SH"])
stream.connect()

The .free() factory method creates a client configured for the free tier API server (https://free-api.tickflow.org). The free tier does not require an API key and is useful for accessing historical daily K-line data, instrument metadata, exchange info, and universe queries.

Limitations of Free Tier:

• No real-time quotes or minute-level K-line data.
• IP-based rate limiting (default: 60 requests/minute).
• Daily K-line data is historical and does not update during trading hours.

from tickflow import TickFlow, AsyncTickFlow

# Synchronous free tier
client = TickFlow.free()
df = client.klines.get("600000.SH", as_dataframe=True)

# Asynchronous free tier
async def main():
    async with AsyncTickFlow.free() as client:
        df = await client.klines.get("600000.SH", as_dataframe=True)

import asyncio
asyncio.run(main())

For high-concurrency applications, use AsyncTickFlow. It is used as an asynchronous context manager.

Free Service:

async with AsyncTickFlow.free() as tf:
    df = await tf.klines.get("600000.SH", period="1d")

Full Service:

async with AsyncTickFlow(api_key="your-api-key") as tf:
    # You can use asyncio.gather to fetch multiple symbols concurrently
    tasks = [tf.klines.get(s) for s in ["600000.SH", "000001.SZ"]]
    results = await asyncio.gather(*tasks)

import asyncio
from tickflow import AsyncTickFlow

async def main():
    async with AsyncTickFlow(api_key="your-api-key") as tf:
        df = await tf.klines.get("600000.SH", as_dataframe=True)
        print(df.tail())

asyncio.run(main())

You can retrieve real-time market quotes using tf.quotes.get().

Query by Symbols

Pass a list of formatted symbol strings (e.g., ["600000.SH", "AAPL.US"]).

Query by Universes (Symbol Pools)

Pass a list of universe identifiers to universes to get quotes for an entire market segment.

• CN_Equity_A: All A-shares
• CN_ETF: All Shenzhen/Shanghai ETFs
• US_Equity: US Equities
• HK_Equity: HK Equities

from tickflow import TickFlow
tf = TickFlow(api_key="your-api-key")

# By specific symbols
quotes = tf.quotes.get(symbols=["600000.SH", "000001.SZ"], as_dataframe=True)

# By universe (e.g., all A-shares)
quotes_a = tf.quotes.get(universes=["CN_Equity_A"], as_dataframe=True)

The TickFlow clients can be configured via direct arguments or environment variables:

• API Key: Pass api_key to the client constructor or set the TICKFLOW_API_KEY environment variable.
• Base URL: Pass base_url to the client constructor or set the TICKFLOW_BASE_URL environment variable.

Important: If no API key is provided (via argument or environment variable), the client will raise a ValueError if it attempts to connect to the default paid URL (https://api.tickflow.org).