Spaces:
Runtime error
Runtime error
File size: 3,035 Bytes
d5b7ee9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """Asset autocomplete widget with symbol and company name search."""
from __future__ import annotations
import logging
import time
import threading
from typing import TYPE_CHECKING
from textual.widgets import Input
from textual_autocomplete import AutoComplete, DropdownItem
from textual_autocomplete._autocomplete import TargetState
if TYPE_CHECKING:
from trading_cli.data.asset_search import AssetSearchEngine
logger = logging.getLogger(__name__)
def create_asset_autocomplete(
search_engine: AssetSearchEngine,
*,
placeholder: str = "Search symbol or company name...",
id: str | None = None, # noqa: A002
) -> tuple[Input, AutoComplete]:
"""Create an Input widget with autocomplete for asset search.
Args:
search_engine: The asset search engine instance.
placeholder: Placeholder text for the input.
id: Widget ID.
Returns:
Tuple of (Input widget, AutoComplete widget).
Yield both in your compose() method.
Example:
input_widget, autocomplete_widget = create_asset_autocomplete(engine)
yield input_widget
yield autocomplete_widget
"""
input_widget = Input(placeholder=placeholder, id=id)
# Cache results to avoid repeated searches
_cache: dict[str, list[DropdownItem]] = {}
_cache_lock = threading.Lock()
_last_query = ""
_last_time = 0.0
def get_suggestions(state: TargetState) -> list[DropdownItem]:
nonlocal _last_query, _last_time
query = state.text.strip()
if not query:
return []
# Debounce: skip if same query within 300ms
now = time.monotonic()
if query == _last_query and (now - _last_time) < 0.3:
return []
_last_query = query
_last_time = now
# Check cache first
with _cache_lock:
if query in _cache:
return _cache[query]
try:
results = search_engine.search(query, max_results=10)
if not results:
return []
suggestions = []
for result in results:
symbol = result["symbol"]
name = result.get("name", "")
# Display format: "AAPL — Apple Inc."
display_text = f"{symbol} — {name}" if name else symbol
suggestions.append(DropdownItem(main=display_text))
# Cache the results
with _cache_lock:
_cache[query] = suggestions
# Limit cache size
if len(_cache) > 1000:
# Remove oldest 500 entries
keys_to_remove = list(_cache.keys())[:500]
for k in keys_to_remove:
del _cache[k]
return suggestions
except Exception as exc:
logger.warning("Asset search failed: %s", exc)
return []
autocomplete = AutoComplete(input_widget, candidates=get_suggestions)
return input_widget, autocomplete
|