pywry.tvchart.udf¶

UDF (Universal Datafeed) adapter that connects a PyWry TradingView chart to any UDF-compatible HTTP server. Handles config discovery, symbol search, bar fetching, marks, quotes, and optional real-time polling.

UDFAdapter¶

pywry.tvchart.udf.UDFAdapter ¶

UDFAdapter(base_url: str, headers: dict[str, str] | None = None, poll_interval: float | None = None, quote_interval: float | None = None, timeout: float = 30.0)

Bases: DatafeedProvider

Connect a PyWry TradingView chart to a UDF HTTP server.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Root URL of the UDF server (e.g. `"https://demo-feed-data.tradingview.com"`).	required
`headers`	`dict or None`	Additional HTTP headers (e.g. `{"Authorization": "Bearer ..."}`) sent with every request.	`None`
`poll_interval`	`float or None`	If set, poll `/history` every poll_interval seconds for the latest bar and push updates via `respond_tvchart_bar_update`. Set to `None` (default) to disable polling.	`None`
`quote_interval`	`float or None`	If set, poll `/quotes` every quote_interval seconds and invoke the `on_quote` callback. Requires symbols to be registered via :meth:`subscribe_quotes`.	`None`
`timeout`	`float`	HTTP request timeout in seconds (default 30).	`30.0`

Attributes¶

_base_url `instance-attribute` ¶

_base_url = rstrip('/')

_headers `instance-attribute` ¶

_headers = headers or {}

_poll_interval `instance-attribute` ¶

_poll_interval = poll_interval

_quote_interval `instance-attribute` ¶

_quote_interval = quote_interval

_timeout `instance-attribute` ¶

_timeout = timeout

_client `instance-attribute` ¶

_client = AsyncClient(base_url=_base_url, headers=_headers, timeout=_timeout)

_config `instance-attribute` ¶

_config: dict[str, Any] | None = None

_supports_search `instance-attribute` ¶

_supports_search: bool = True

_supports_group_request `instance-attribute` ¶

_supports_group_request: bool = False

_supports_marks `instance-attribute` ¶

_supports_marks: bool = False

_supports_timescale_marks `instance-attribute` ¶

_supports_timescale_marks: bool = False

_supports_time `instance-attribute` ¶

_supports_time: bool = False

_subscriptions `instance-attribute` ¶

_subscriptions: dict[str, dict[str, Any]] = {}

_poll_timers `instance-attribute` ¶

_poll_timers: dict[str, Timer] = {}

_quote_symbols `instance-attribute` ¶

_quote_symbols: set[str] = set()

_quote_timer `instance-attribute` ¶

_quote_timer: Timer | None = None

_on_quote `instance-attribute` ¶

_on_quote: Any = None

_app `instance-attribute` ¶

_app: PyWry | None = None

_closed `instance-attribute` ¶

_closed = False

supports_marks `property` ¶

supports_marks: bool

Whether the UDF server supports chart marks.

supports_timescale_marks `property` ¶

supports_timescale_marks: bool

Whether the UDF server supports timescale marks.

supports_time `property` ¶

supports_time: bool

Whether the UDF server supports the /time endpoint.

supports_search `property` ¶

supports_search: bool

Whether the UDF server supports symbol search.

config `property` ¶

config: dict[str, Any] | None

The cached UDF server configuration, or None if not yet fetched.

Functions¶

get_config `async` ¶

get_config() -> dict[str, Any]

Fetch and cache /config from the UDF server.

search_symbols `async` ¶

search_symbols(query: str, symbol_type: str = '', exchange: str = '', limit: int = 30) -> list[dict[str, Any]]

Search UDF server for matching symbols.

resolve_symbol `async` ¶

resolve_symbol(symbol: str) -> dict[str, Any]

Resolve symbol metadata from the UDF server.

get_bars `async` ¶

get_bars(symbol: str, resolution: str, from_ts: int, to_ts: int, countback: int | None = None) -> dict[str, Any]

Fetch bars from /history and convert to the PyWry response format.

get_marks `async` ¶

get_marks(symbol: str, from_ts: int, to_ts: int, resolution: str) -> list[dict[str, Any]]

Fetch chart marks from the UDF /marks endpoint.

get_timescale_marks `async` ¶

get_timescale_marks(symbol: str, from_ts: int, to_ts: int, resolution: str) -> list[dict[str, Any]]

Fetch timescale marks from the UDF /timescale_marks endpoint.

get_server_time `async` ¶

get_server_time() -> int

Fetch server time from the UDF /time endpoint.

on_subscribe ¶

on_subscribe(listener_guid: str, symbol: str, resolution: str, chart_id: str | None = None) -> None

Track a bar subscription and start polling if configured.

on_unsubscribe ¶

on_unsubscribe(listener_guid: str) -> None

Remove a bar subscription and stop its poll timer.

_fetch_config `async` ¶

_fetch_config() -> dict[str, Any]

_get_quotes `async` ¶

_get_quotes(symbols: list[str]) -> list[QuoteData]

Fetch real-time quotes for one or more symbols.

subscribe_quotes ¶

subscribe_quotes(symbols: list[str], on_quote: Any = None) -> None

Register symbols for periodic quote polling.

Parameters:

Name	Type	Description	Default
`symbols`	`list[str]`	Symbols to track (e.g. `["NYSE:AA", "NASDAQ:AAPL"]`).	required
`on_quote`	`callable`	`fn(quotes: list[QuoteData])` called on each poll cycle. If not set, previously registered callback is kept.	`None`

unsubscribe_quotes ¶

unsubscribe_quotes(symbols: list[str] | None = None) -> None

Remove symbols from quote polling.

Parameters:

Name	Type	Description	Default
`symbols`	`list[str] or None`	Symbols to remove. If None, removes all.	`None`

connect ¶

connect(app: PyWry, symbol: str = 'AAPL', resolution: str = 'D', **show_kwargs: Any) -> Any

Wire up all UDF datafeed events and show the chart.

This is the main entry point. Call it once; it fetches /config from the UDF server, auto-wires all datafeed event handlers via the :class:DatafeedProvider protocol, and calls app.show_tvchart(use_datafeed=True).

Parameters:

Name	Type	Description	Default
`app`	`PyWry`	The application instance.	required
`symbol`	`str`	Initial symbol to display (default `"AAPL"`).	`'AAPL'`
`resolution`	`str`	Initial resolution in UDF format (default `"D"`).	`'D'`
`**show_kwargs`	`Any`	Extra keyword arguments forwarded to `app.show_tvchart()`, e.g. `title`, `width`, `height`, `toolbars`, `chart_options`.	`{}`

Returns:

Type	Description
`NativeWindowHandle or BaseWidget`	The window or widget returned by `show_tvchart`.

_start_bar_poll ¶

_start_bar_poll(listener_guid: str) -> None

_stop_bar_poll ¶

_stop_bar_poll(listener_guid: str) -> None

_start_quote_polling ¶

_start_quote_polling() -> None

_stop_quote_polling ¶

_stop_quote_polling() -> None

close ¶

close() -> None

Shut down the adapter: cancel timers and close the HTTP client.

QuoteData¶