Spaces:

igardner
/

StockAnalysis

Running

feat: make stock analysis the primary Gradio app

cf05d66 about 22 hours ago

20.5 kB

"""StockAnalysis - JSON interface for an external stock-analysis project.
The main app is intentionally thin. It collects a ticker and a few optional
inputs, sends them to the external project through ``external_api.py``, and
returns JSON plus the raw upstream body. If the upstream returns text and a
local OpenAI-compatible model is configured, ``local_parser.py`` converts it to
JSON without inventing new facts.
The original Hugging Face inference smoke test is still available in a compact
panel fixed to the bottom-right corner. Its Gradio events are marked private so
they are not advertised in the public API schema.
Settings live in the SETTINGS block below. ``HF_TOKEN`` is read from the Space
secret through ``os.getenv`` and is never written to disk.
"""

from __future__ import annotations

import html
import os
from pathlib import Path
from typing import Any

import gradio as gr

import external_api
import local_parser
import smoke

# =============================================================================
# SETTINGS - edit these, then restart the Space.
# =============================================================================

# External stock-analysis project. Leave blank until the project is ready.
EXTERNAL_API_URL = os.getenv("EXTERNAL_API_URL", "")
EXTERNAL_API_TIMEOUT_SECONDS = 120

# Uploaded custom model ID. Leave blank until the model is available.
STOCK_MODEL_ID = os.getenv("STOCK_MODEL_ID", "")

# Optional local parser. Any OpenAI-compatible endpoint works.
LOCAL_PARSER_URL = os.getenv("LOCAL_PARSER_URL", "")
LOCAL_PARSER_MODEL = os.getenv("LOCAL_PARSER_MODEL", "llama3.1")
LOCAL_PARSER_TIMEOUT_SECONDS = 90

# Stock-analysis form choices.
HORIZONS = ["1 week", "1 month", "3 months", "6 months", "1 year"]
RISK_PROFILES = ["conservative", "balanced", "aggressive"]

# Inference smoke test settings (secondary feature).
MODEL_LIMIT = 30
CONCURRENCY = 3
MAX_TOKENS = 8
REQUEST_TIMEOUT_SECONDS = 60
RESULTS_DIR = "/data/hf-inference-smoke"
PAGE_SIZE = 100
MAX_SCAN = 500
AUTO_RUN_ON_START = False

# =============================================================================

smoke.configure(
    results_root=RESULTS_DIR,
    model_limit=MODEL_LIMIT,
    concurrency=CONCURRENCY,
    max_tokens=MAX_TOKENS,
    request_timeout=REQUEST_TIMEOUT_SECONDS,
    page_size=PAGE_SIZE,
    max_scan=MAX_SCAN,
)

external_api.configure(
    url=EXTERNAL_API_URL,
    timeout=EXTERNAL_API_TIMEOUT_SECONDS,
    model_id=STOCK_MODEL_ID,
)

local_parser.configure(
    url=LOCAL_PARSER_URL,
    model=LOCAL_PARSER_MODEL,
    timeout=LOCAL_PARSER_TIMEOUT_SECONDS,
)


def _space_host() -> str:
    raw = os.getenv("SPACE_HOST", "igardner-stockanalysis.hf.space").strip()
    raw = raw.removeprefix("https://").removeprefix("http://").strip("/")
    return raw or "igardner-stockanalysis.hf.space"


SPACE_ID = os.getenv("SPACE_ID", "igardner/StockAnalysis").strip() or "igardner/StockAnalysis"
SPACE_HOST = _space_host()
AGENTS_PATH = Path(__file__).with_name("agents.md")

SMOKE_TABLE_HEADERS = ["Model", "Passed", "Latency (ms)", "Error"]
SMOKE_TABLE_TYPES = ["str", "bool", "number", "str"]

SETTINGS_MD = f"""
| Setting | Value |
|---|---|
| Models tested | **{MODEL_LIMIT}** trending with a live provider |
| Concurrency | {CONCURRENCY} |
| Max output tokens | {MAX_TOKENS} per attempt |
| Request timeout | {REQUEST_TIMEOUT_SECONDS}s |
| Routing | `:cheapest`, then one `:fastest` retry |
| Results directory | `{RESULTS_DIR}` |
| Auto-run on start | {AUTO_RUN_ON_START} |
"""

HOW_IT_WORKS = """
### How the smoke test works
1. Queries the Hub for `conversational` models sorted by `trendingScore` and
   skips models without a live inference provider.
2. Calls each model through the HF router with `:cheapest`, then retries once
   with `:fastest` if the first attempt fails.
3. Marks a model as passed only when the router returns HTTP 200 with a
   non-empty `choices` array.
4. Saves every run to `/data/hf-inference-smoke/runs/` and mirrors the latest
   run to `latest.json`.
"""

HERO_MD = """
# StockAnalysis
Structured stock-analysis output for humans and agents.
Enter a ticker, choose a horizon and risk profile, and the Space returns the
external project's response as JSON. The raw upstream body is kept beside it.
If the upstream returns text, an optional local model can convert it to JSON.
The agent contract is the normal Gradio API. See **API / agents** below or read
`agents.md`.
"""

DISCLAIMER_MD = """
> **Research tooling, not financial advice.** Output comes from an external
> project and may be incomplete or wrong. No return is guaranteed, and you can
> lose money. Verify anything before acting on it.
"""


def _esc(value: Any) -> str:
    return html.escape(str(value), quote=True)


def _pill(label: str, ok: bool, detail: str) -> str:
    state = "ok" if ok else "warn"
    text = f"{label}: {detail}"
    return (
        f'<span class="status-pill {state}">'
        f'<span class="status-dot"></span>{_esc(text)}</span>'
    )


def _status_html() -> str:
    external_ok = external_api.is_configured()
    parser_ok = local_parser.is_configured()
    model_ok = bool(external_api.model_id())
    token_ok = bool(os.getenv("HF_TOKEN"))
    pills = [
        _pill("External API", external_ok, "configured" if external_ok else "not configured"),
        _pill("Local parser", parser_ok, local_parser.model_name() if parser_ok else "off"),
        _pill("Custom model", model_ok, external_api.model_id() if model_ok else "not set"),
        _pill("HF token", token_ok, "configured" if token_ok else "missing"),
    ]
    return '<div class="status-row">' + "".join(pills) + "</div>"


def _api_md() -> str:
    return f"""
### Agent contract
There is no separate REST layer. Agents call the normal Gradio event behind the
**Analyze** button.
- Schema: `GET https://{SPACE_HOST}/gradio_api/info`
- Call: `POST https://{SPACE_HOST}/gradio_api/call/v2/analyze`
- Poll: `GET https://{SPACE_HOST}/gradio_api/call/analyze/{{event_id}}`
- Auth: `Bearer $HF_TOKEN` (https://huggingface.co/settings/tokens)
- File inputs: none.
The Hub also serves a generated live `agents.md` at
`https://huggingface.co/spaces/{SPACE_ID}/agents.md`. The repository
`agents.md` contains the concrete request and response examples.
"""


def _token_banner() -> str:
    if os.getenv("HF_TOKEN"):
        return "**Token status:** `HF_TOKEN` is configured."
    return (
        "**Token status:** `HF_TOKEN` is not set. "
        "Add it as a Space secret to run the smoke test."
    )


def _summary(run: dict[str, Any] | None) -> str:
    if not run:
        return "No runs yet. Press **Run smoke test**."
    total = run.get("model_limit") or len(run.get("results", []))
    passed = run.get("passed", 0)
    failed = run.get("failed", 0)
    duration = run.get("duration_seconds", 0)
    return (
        f"**Run `{run.get('run_id')}`** - **{passed}/{total}** passed "
        f"in {duration}s ({failed} failed)."
    )


def _rows(run: dict[str, Any] | None) -> list[list[Any]]:
    if not run:
        return []
    return [
        [
            item.get("model"),
            bool(item.get("ok")),
            item.get("latency_ms"),
            item.get("error") or "",
        ]
        for item in run.get("results", [])
    ]


def _run_options() -> list[tuple[str, str]]:
    options: list[tuple[str, str]] = []
    for meta in smoke.list_runs():
        label = (
            f"{meta.get('run_id')} - {meta.get('passed')}/"
            f"{meta.get('model_limit')} passed"
        )
        options.append((label, meta.get("run_id")))
    return options


def _download_path(run: dict[str, Any] | None) -> str | None:
    if not run:
        return None
    run_id = run.get("run_id")
    if not run_id:
        return None
    path = smoke.RUNS_DIR / f"{run_id}.json"
    return str(path) if path.exists() else None


def _view(
    run: dict[str, Any] | None,
) -> tuple[str, list[list[Any]], str | None]:
    return _summary(run), _rows(run), _download_path(run)


def _agents_text() -> str:
    try:
        return AGENTS_PATH.read_text(encoding="utf-8")
    except OSError:
        return ""


def analyze_stock(
    ticker: str,
    horizon: str = "3 months",
    risk_profile: str = "balanced",
    question: str = "",
) -> tuple[Any, str, str]:
    """Return stock-analysis results as JSON.
    Args:
        ticker: Stock ticker symbol, for example AAPL or BTC-USD.
        horizon: Requested analysis horizon, for example 3 months.
        risk_profile: One of conservative, balanced, or aggressive.
        question: Optional free-form question for the external project.
    """
    clean_ticker = (ticker or "").strip().upper()
    if not clean_ticker:
        return (
            {"status": "invalid_input", "error": "ticker is required"},
            "",
            "Enter a ticker symbol.",
        )
    if len(clean_ticker) > 24:
        return (
            {
                "status": "invalid_input",
                "error": "ticker must be 24 characters or fewer",
            },
            "",
            "Ticker is too long.",
        )

    if horizon not in HORIZONS:
        horizon = "3 months"
    if risk_profile not in RISK_PROFILES:
        risk_profile = "balanced"

    if not external_api.is_configured():
        return (
            {
                "status": "not_configured",
                "ticker": clean_ticker,
                "message": "Set EXTERNAL_API_URL to the stock-analysis project endpoint.",
            },
            "",
            "External project is not configured. Set `EXTERNAL_API_URL`.",
        )

    result = external_api.query_external(
        ticker=clean_ticker,
        horizon=horizon,
        risk_profile=risk_profile,
        question=(question or "").strip()[:2000],
    )

    if result.error and not result.is_json:
        return (
            result.payload,
            result.raw,
            "External request failed. The JSON result has the details.",
        )

    if result.is_json:
        return (
            result.payload,
            result.raw,
            f"Returned JSON from the external project (HTTP {result.http_status}).",
        )

    if local_parser.is_configured():
        try:
            parsed = local_parser.parse_json(result.raw, ticker=clean_ticker)
        except local_parser.LocalParserError as exc:
            payload = result.payload
            if isinstance(payload, dict):
                payload = {**payload, "parse_error": str(exc)}
            return payload, result.raw, f"Local parser failed: {exc}"

        payload = {
            "status": "ok",
            "ticker": clean_ticker,
            "source": "external+local_parser",
            "http_status": result.http_status,
            "content_type": result.content_type,
            "parser": {"model": local_parser.model_name()},
            "analysis": parsed,
            "raw": result.raw,
        }
        return (
            payload,
            result.raw,
            f"Converted the raw response to JSON with `{local_parser.model_name()}`.",
        )

    return (
        result.payload,
        result.raw,
        "Returned the raw upstream response. Set `LOCAL_PARSER_URL` to convert it to JSON.",
    )


def initial_view() -> tuple[str, list[list[Any]], str | None, gr.Dropdown, str]:
    run = smoke.load_latest_run()
    summary, rows, download = _view(run)
    choices = _run_options()
    selected = run.get("run_id") if run else None
    return (
        summary,
        rows,
        download,
        gr.Dropdown(choices=choices, value=selected),
        _token_banner(),
    )


def run_smoke_test(
    progress: gr.Progress = gr.Progress(),
) -> tuple[str, list[list[Any]], str | None, gr.Dropdown, str]:
    token = os.getenv("HF_TOKEN")
    if not token:
        raise gr.Error(
            "HF_TOKEN is not set. Add it as a Space secret or environment variable, "
            "then restart the Space."
        )

    status = smoke.get_token_status(token)
    if not status["ok"]:
        raise gr.Error(f"HF_TOKEN was rejected: {status['error']}")

    progress(0.0, desc="Fetching trending models...")
    try:
        snapshot = smoke.fetch_trending_models(limit=MODEL_LIMIT)
    except Exception as exc:  # noqa: BLE001 - surface discovery failures in the UI
        raise gr.Error(f"Model discovery failed: {exc}") from exc

    models = snapshot["models"]
    total = len(models)

    def on_result(done: int, count: int, result: dict[str, Any]) -> None:
        state = "PASS" if result["ok"] else "FAIL"
        progress(
            done / count,
            desc=f"{done}/{count} - {state} {result['model']} ({result['attempt']})",
        )

    progress(0.0, desc=f"Testing {total} models...")
    try:
        run = smoke.run_smoke(
            models,
            token=token,
            concurrency=CONCURRENCY,
            timeout=REQUEST_TIMEOUT_SECONDS,
            progress=on_result,
        )
    except Exception as exc:  # noqa: BLE001 - surface runner failures in the UI
        raise gr.Error(f"Smoke test failed: {exc}") from exc

    summary, rows, download = _view(run)
    choices = _run_options()
    banner = _token_banner()
    return (
        summary,
        rows,
        download,
        gr.Dropdown(choices=choices, value=run["run_id"]),
        banner,
    )


def maybe_auto_run(
    progress: gr.Progress = gr.Progress(),
) -> tuple[str, list[list[Any]], str | None, gr.Dropdown, str]:
    """Run at startup only when enabled in SETTINGS and a token is available."""
    if AUTO_RUN_ON_START and os.getenv("HF_TOKEN"):
        return run_smoke_test(progress=progress)
    return initial_view()


def load_history(run_id: str | None) -> tuple[str, list[list[Any]], str | None]:
    if not run_id:
        run = smoke.load_latest_run()
    else:
        try:
            run = smoke.load_run(run_id)
        except ValueError as exc:
            raise gr.Error(str(exc)) from exc
    if run is None:
        raise gr.Error("That run could not be found in storage.")
    return _view(run)


CSS = """
#smoke-corner {
  position: fixed;
  right: 16px;
  bottom: 16px;
  width: 380px;
  max-width: calc(100vw - 32px);
  max-height: 70vh;
  overflow-y: auto;
  z-index: 1000;
  padding: 6px;
  border: 1px solid var(--border-color-primary);
  border-radius: 12px;
  background: var(--background-fill-primary);
  box-shadow: 0 12px 32px rgba(0, 0, 0, 0.18);
}
#smoke-corner .block {
  margin: 0;
}
#smoke-corner .label-wrap {
  padding: 4px 8px;
}
#smoke-corner .prose {
  font-size: 0.85rem;
}
#smoke-corner table {
  font-size: 0.78rem;
}
.status-row {
  display: flex;
  flex-wrap: wrap;
  gap: 8px;
  margin: 8px 0 16px;
}
.status-pill {
  display: inline-flex;
  align-items: center;
  gap: 6px;
  padding: 4px 10px;
  border-radius: 999px;
  border: 1px solid var(--border-color-primary);
  background: var(--background-fill-secondary);
  font-size: 0.8rem;
  line-height: 1.4;
}
.status-pill.ok {
  color: #0a7f3f;
}
.status-pill.warn {
  color: #b45309;
}
.status-dot {
  width: 8px;
  height: 8px;
  border-radius: 50%;
  background: currentColor;
  flex: 0 0 auto;
}
@media (max-width: 640px) {
  #smoke-corner {
    left: 10px;
    right: 10px;
    width: auto;
    max-height: 60vh;
  }
}
"""


with gr.Blocks(title="StockAnalysis") as demo:
    gr.Markdown(HERO_MD)
    gr.HTML(_status_html())

    with gr.Row():
        with gr.Column(scale=1, min_width=280):
            ticker = gr.Textbox(
                label="Ticker",
                value="AAPL",
                placeholder="AAPL, MSFT, NVDA, BTC-USD",
                info="Symbol sent to the external project.",
            )
            horizon = gr.Dropdown(
                choices=HORIZONS,
                value="3 months",
                label="Horizon",
            )
            risk_profile = gr.Dropdown(
                choices=RISK_PROFILES,
                value="balanced",
                label="Risk profile",
            )
            question = gr.Textbox(
                label="Question (optional)",
                lines=3,
                placeholder="What should the analysis focus on?",
            )
            analyze_button = gr.Button("Analyze", variant="primary")
            gr.Markdown(DISCLAIMER_MD)

        with gr.Column(scale=2):
            result_status = gr.Markdown("*Ready.*")
            json_output = gr.JSON(
                label="JSON result",
                open=True,
                max_height=520,
                show_indices=False,
            )
            with gr.Accordion("Raw upstream response", open=False):
                raw_output = gr.Textbox(
                    label="Raw response",
                    lines=12,
                    max_lines=20,
                    interactive=False,
                    buttons=["copy"],
                )

    with gr.Accordion("API / agents", open=False):
        gr.Markdown(_api_md())
        agents_text = _agents_text()
        if agents_text:
            gr.Code(
                agents_text,
                language="markdown",
                label="agents.md",
                lines=16,
                max_lines=30,
                wrap_lines=True,
                show_line_numbers=False,
            )
            if AGENTS_PATH.exists():
                gr.DownloadButton(
                    "Download agents.md",
                    value=str(AGENTS_PATH),
                    size="sm",
                )
        else:
            gr.Markdown("*The repository `agents.md` is not present.*")

    with gr.Column(elem_id="smoke-corner"):
        with gr.Accordion("Inference smoke test", open=False):
            smoke_token_banner = gr.Markdown(_token_banner())
            smoke_run_button = gr.Button(
                "Run smoke test",
                size="sm",
                variant="secondary",
            )
            smoke_progress_note = gr.Markdown("*Idle.*")
            smoke_summary = gr.Markdown(_summary(None))
            smoke_table = gr.Dataframe(
                headers=SMOKE_TABLE_HEADERS,
                datatype=SMOKE_TABLE_TYPES,
                value=[],
                interactive=False,
                wrap=True,
                max_height=170,
                show_row_numbers=False,
                show_search=False,
                label="Models used (True = valid chat completion)",
            )
            with gr.Accordion("History & settings", open=False):
                smoke_history = gr.Dropdown(
                    choices=[],
                    value=None,
                    label="Run history",
                    info=f"Load a previous run from {RESULTS_DIR}/runs.",
                )
                smoke_download = gr.File(
                    label="Download run JSON",
                    interactive=False,
                )
                gr.Markdown(SETTINGS_MD)
                gr.Markdown(HOW_IT_WORKS)

    analyze_button.click(
        fn=analyze_stock,
        inputs=[ticker, horizon, risk_profile, question],
        outputs=[json_output, raw_output, result_status],
        api_name="analyze",
        api_description=(
            "Return the external stock-analysis project's response as JSON, "
            "the raw upstream body, and a short status string."
        ),
        api_visibility="public",
        concurrency_limit=4,
        concurrency_id="analyze",
    )

    smoke_run_button.click(
        fn=run_smoke_test,
        inputs=[],
        outputs=[
            smoke_summary,
            smoke_table,
            smoke_download,
            smoke_history,
            smoke_token_banner,
        ],
        api_visibility="private",
        concurrency_limit=1,
        concurrency_id="smoke",
    ).success(
        lambda: "*Run finished.*",
        outputs=smoke_progress_note,
        api_visibility="private",
    )

    smoke_history.change(
        fn=load_history,
        inputs=[smoke_history],
        outputs=[smoke_summary, smoke_table, smoke_download],
        api_visibility="private",
    )

    demo.load(
        fn=maybe_auto_run,
        outputs=[
            smoke_summary,
            smoke_table,
            smoke_download,
            smoke_history,
            smoke_token_banner,
        ],
        api_visibility="private",
    )


if __name__ == "__main__":
    demo.launch(
        theme=gr.themes.Soft(),
        css=CSS,
        allowed_paths=[RESULTS_DIR],
    )

	"""StockAnalysis - JSON interface for an external stock-analysis project.

	The main app is intentionally thin. It collects a ticker and a few optional
	inputs, sends them to the external project through ``external_api.py``, and
	returns JSON plus the raw upstream body. If the upstream returns text and a
	local OpenAI-compatible model is configured, ``local_parser.py`` converts it to
	JSON without inventing new facts.

	The original Hugging Face inference smoke test is still available in a compact
	panel fixed to the bottom-right corner. Its Gradio events are marked private so
	they are not advertised in the public API schema.

	Settings live in the SETTINGS block below. ``HF_TOKEN`` is read from the Space
	secret through ``os.getenv`` and is never written to disk.
	"""

	from __future__ import annotations

	import html
	import os
	from pathlib import Path
	from typing import Any

	import gradio as gr

	import external_api
	import local_parser
	import smoke

	# =============================================================================
	# SETTINGS - edit these, then restart the Space.
	# =============================================================================

	# External stock-analysis project. Leave blank until the project is ready.
	EXTERNAL_API_URL = os.getenv("EXTERNAL_API_URL", "")
	EXTERNAL_API_TIMEOUT_SECONDS = 120

	# Uploaded custom model ID. Leave blank until the model is available.
	STOCK_MODEL_ID = os.getenv("STOCK_MODEL_ID", "")

	# Optional local parser. Any OpenAI-compatible endpoint works.
	LOCAL_PARSER_URL = os.getenv("LOCAL_PARSER_URL", "")
	LOCAL_PARSER_MODEL = os.getenv("LOCAL_PARSER_MODEL", "llama3.1")
	LOCAL_PARSER_TIMEOUT_SECONDS = 90

	# Stock-analysis form choices.
	HORIZONS = ["1 week", "1 month", "3 months", "6 months", "1 year"]
	RISK_PROFILES = ["conservative", "balanced", "aggressive"]

	# Inference smoke test settings (secondary feature).
	MODEL_LIMIT = 30
	CONCURRENCY = 3
	MAX_TOKENS = 8
	REQUEST_TIMEOUT_SECONDS = 60
	RESULTS_DIR = "/data/hf-inference-smoke"
	PAGE_SIZE = 100
	MAX_SCAN = 500
	AUTO_RUN_ON_START = False

	# =============================================================================

	smoke.configure(
	results_root=RESULTS_DIR,
	model_limit=MODEL_LIMIT,
	concurrency=CONCURRENCY,
	max_tokens=MAX_TOKENS,
	request_timeout=REQUEST_TIMEOUT_SECONDS,
	page_size=PAGE_SIZE,
	max_scan=MAX_SCAN,
	)

	external_api.configure(
	url=EXTERNAL_API_URL,
	timeout=EXTERNAL_API_TIMEOUT_SECONDS,
	model_id=STOCK_MODEL_ID,
	)

	local_parser.configure(
	url=LOCAL_PARSER_URL,
	model=LOCAL_PARSER_MODEL,
	timeout=LOCAL_PARSER_TIMEOUT_SECONDS,
	)


	def _space_host() -> str:
	raw = os.getenv("SPACE_HOST", "igardner-stockanalysis.hf.space").strip()
	raw = raw.removeprefix("https://").removeprefix("http://").strip("/")
	return raw or "igardner-stockanalysis.hf.space"


	SPACE_ID = os.getenv("SPACE_ID", "igardner/StockAnalysis").strip() or "igardner/StockAnalysis"
	SPACE_HOST = _space_host()
	AGENTS_PATH = Path(__file__).with_name("agents.md")

	SMOKE_TABLE_HEADERS = ["Model", "Passed", "Latency (ms)", "Error"]
	SMOKE_TABLE_TYPES = ["str", "bool", "number", "str"]

	SETTINGS_MD = f"""
	\| Setting \| Value \|
	\|---\|---\|
	\| Models tested \| {MODEL_LIMIT} trending with a live provider \|
	\| Concurrency \| {CONCURRENCY} \|
	\| Max output tokens \| {MAX_TOKENS} per attempt \|
	\| Request timeout \| {REQUEST_TIMEOUT_SECONDS}s \|
	\| Routing \| `:cheapest`, then one `:fastest` retry \|
	\| Results directory \| `{RESULTS_DIR}` \|
	\| Auto-run on start \| {AUTO_RUN_ON_START} \|
	"""

	HOW_IT_WORKS = """
	### How the smoke test works

	1. Queries the Hub for `conversational` models sorted by `trendingScore` and
	skips models without a live inference provider.
	2. Calls each model through the HF router with `:cheapest`, then retries once
	with `:fastest` if the first attempt fails.
	3. Marks a model as passed only when the router returns HTTP 200 with a
	non-empty `choices` array.
	4. Saves every run to `/data/hf-inference-smoke/runs/` and mirrors the latest
	run to `latest.json`.
	"""

	HERO_MD = """
	# StockAnalysis

	Structured stock-analysis output for humans and agents.

	Enter a ticker, choose a horizon and risk profile, and the Space returns the
	external project's response as JSON. The raw upstream body is kept beside it.
	If the upstream returns text, an optional local model can convert it to JSON.

	The agent contract is the normal Gradio API. See API / agents below or read
	`agents.md`.
	"""

	DISCLAIMER_MD = """
	> Research tooling, not financial advice. Output comes from an external
	> project and may be incomplete or wrong. No return is guaranteed, and you can
	> lose money. Verify anything before acting on it.
	"""


	def _esc(value: Any) -> str:
	return html.escape(str(value), quote=True)


	def _pill(label: str, ok: bool, detail: str) -> str:
	state = "ok" if ok else "warn"
	text = f"{label}: {detail}"
	return (
	f'<span class="status-pill {state}">'
	f'<span class="status-dot"></span>{_esc(text)}</span>'
	)


	def _status_html() -> str:
	external_ok = external_api.is_configured()
	parser_ok = local_parser.is_configured()
	model_ok = bool(external_api.model_id())
	token_ok = bool(os.getenv("HF_TOKEN"))
	pills = [
	_pill("External API", external_ok, "configured" if external_ok else "not configured"),
	_pill("Local parser", parser_ok, local_parser.model_name() if parser_ok else "off"),
	_pill("Custom model", model_ok, external_api.model_id() if model_ok else "not set"),
	_pill("HF token", token_ok, "configured" if token_ok else "missing"),
	]
	return '<div class="status-row">' + "".join(pills) + "</div>"


	def _api_md() -> str:
	return f"""
	### Agent contract

	There is no separate REST layer. Agents call the normal Gradio event behind the
	Analyze button.

	- Schema: `GET https://{SPACE_HOST}/gradio_api/info`
	- Call: `POST https://{SPACE_HOST}/gradio_api/call/v2/analyze`
	- Poll: `GET https://{SPACE_HOST}/gradio_api/call/analyze/{{event_id}}`
	- Auth: `Bearer $HF_TOKEN` (https://huggingface.co/settings/tokens)
	- File inputs: none.

	The Hub also serves a generated live `agents.md` at
	`https://huggingface.co/spaces/{SPACE_ID}/agents.md`. The repository
	`agents.md` contains the concrete request and response examples.
	"""


	def _token_banner() -> str:
	if os.getenv("HF_TOKEN"):
	return "Token status: `HF_TOKEN` is configured."
	return (
	"Token status: `HF_TOKEN` is not set. "
	"Add it as a Space secret to run the smoke test."
	)


	def _summary(run: dict[str, Any] \| None) -> str:
	if not run:
	return "No runs yet. Press Run smoke test."
	total = run.get("model_limit") or len(run.get("results", []))
	passed = run.get("passed", 0)
	failed = run.get("failed", 0)
	duration = run.get("duration_seconds", 0)
	return (
	f"Run `{run.get('run_id')}` - {passed}/{total} passed "
	f"in {duration}s ({failed} failed)."
	)


	def _rows(run: dict[str, Any] \| None) -> list[list[Any]]:
	if not run:
	return []
	return [
	[
	item.get("model"),
	bool(item.get("ok")),
	item.get("latency_ms"),
	item.get("error") or "",
	]
	for item in run.get("results", [])
	]


	def _run_options() -> list[tuple[str, str]]:
	options: list[tuple[str, str]] = []
	for meta in smoke.list_runs():
	label = (
	f"{meta.get('run_id')} - {meta.get('passed')}/"
	f"{meta.get('model_limit')} passed"
	)
	options.append((label, meta.get("run_id")))
	return options


	def _download_path(run: dict[str, Any] \| None) -> str \| None:
	if not run:
	return None
	run_id = run.get("run_id")
	if not run_id:
	return None
	path = smoke.RUNS_DIR / f"{run_id}.json"
	return str(path) if path.exists() else None


	def _view(
	run: dict[str, Any] \| None,
	) -> tuple[str, list[list[Any]], str \| None]:
	return _summary(run), _rows(run), _download_path(run)


	def _agents_text() -> str:
	try:
	return AGENTS_PATH.read_text(encoding="utf-8")
	except OSError:
	return ""


	def analyze_stock(
	ticker: str,
	horizon: str = "3 months",
	risk_profile: str = "balanced",
	question: str = "",
	) -> tuple[Any, str, str]:
	"""Return stock-analysis results as JSON.

	Args:
	ticker: Stock ticker symbol, for example AAPL or BTC-USD.
	horizon: Requested analysis horizon, for example 3 months.
	risk_profile: One of conservative, balanced, or aggressive.
	question: Optional free-form question for the external project.
	"""
	clean_ticker = (ticker or "").strip().upper()
	if not clean_ticker:
	return (
	{"status": "invalid_input", "error": "ticker is required"},
	"",
	"Enter a ticker symbol.",
	)
	if len(clean_ticker) > 24:
	return (
	{
	"status": "invalid_input",
	"error": "ticker must be 24 characters or fewer",
	},
	"",
	"Ticker is too long.",
	)

	if horizon not in HORIZONS:
	horizon = "3 months"
	if risk_profile not in RISK_PROFILES:
	risk_profile = "balanced"

	if not external_api.is_configured():
	return (
	{
	"status": "not_configured",
	"ticker": clean_ticker,
	"message": "Set EXTERNAL_API_URL to the stock-analysis project endpoint.",
	},
	"",
	"External project is not configured. Set `EXTERNAL_API_URL`.",
	)

	result = external_api.query_external(
	ticker=clean_ticker,
	horizon=horizon,
	risk_profile=risk_profile,
	question=(question or "").strip()[:2000],
	)

	if result.error and not result.is_json:
	return (
	result.payload,
	result.raw,
	"External request failed. The JSON result has the details.",
	)

	if result.is_json:
	return (
	result.payload,
	result.raw,
	f"Returned JSON from the external project (HTTP {result.http_status}).",
	)

	if local_parser.is_configured():
	try:
	parsed = local_parser.parse_json(result.raw, ticker=clean_ticker)
	except local_parser.LocalParserError as exc:
	payload = result.payload
	if isinstance(payload, dict):
	payload = {**payload, "parse_error": str(exc)}
	return payload, result.raw, f"Local parser failed: {exc}"

	payload = {
	"status": "ok",
	"ticker": clean_ticker,
	"source": "external+local_parser",
	"http_status": result.http_status,
	"content_type": result.content_type,
	"parser": {"model": local_parser.model_name()},
	"analysis": parsed,
	"raw": result.raw,
	}
	return (
	payload,
	result.raw,
	f"Converted the raw response to JSON with `{local_parser.model_name()}`.",
	)

	return (
	result.payload,
	result.raw,
	"Returned the raw upstream response. Set `LOCAL_PARSER_URL` to convert it to JSON.",
	)


	def initial_view() -> tuple[str, list[list[Any]], str \| None, gr.Dropdown, str]:
	run = smoke.load_latest_run()
	summary, rows, download = _view(run)
	choices = _run_options()
	selected = run.get("run_id") if run else None
	return (
	summary,
	rows,
	download,
	gr.Dropdown(choices=choices, value=selected),
	_token_banner(),
	)


	def run_smoke_test(
	progress: gr.Progress = gr.Progress(),
	) -> tuple[str, list[list[Any]], str \| None, gr.Dropdown, str]:
	token = os.getenv("HF_TOKEN")
	if not token:
	raise gr.Error(
	"HF_TOKEN is not set. Add it as a Space secret or environment variable, "
	"then restart the Space."
	)

	status = smoke.get_token_status(token)
	if not status["ok"]:
	raise gr.Error(f"HF_TOKEN was rejected: {status['error']}")

	progress(0.0, desc="Fetching trending models...")
	try:
	snapshot = smoke.fetch_trending_models(limit=MODEL_LIMIT)
	except Exception as exc: # noqa: BLE001 - surface discovery failures in the UI
	raise gr.Error(f"Model discovery failed: {exc}") from exc

	models = snapshot["models"]
	total = len(models)

	def on_result(done: int, count: int, result: dict[str, Any]) -> None:
	state = "PASS" if result["ok"] else "FAIL"
	progress(
	done / count,
	desc=f"{done}/{count} - {state} {result['model']} ({result['attempt']})",
	)

	progress(0.0, desc=f"Testing {total} models...")
	try:
	run = smoke.run_smoke(
	models,
	token=token,
	concurrency=CONCURRENCY,
	timeout=REQUEST_TIMEOUT_SECONDS,
	progress=on_result,
	)
	except Exception as exc: # noqa: BLE001 - surface runner failures in the UI
	raise gr.Error(f"Smoke test failed: {exc}") from exc

	summary, rows, download = _view(run)
	choices = _run_options()
	banner = _token_banner()
	return (
	summary,
	rows,
	download,
	gr.Dropdown(choices=choices, value=run["run_id"]),
	banner,
	)


	def maybe_auto_run(
	progress: gr.Progress = gr.Progress(),
	) -> tuple[str, list[list[Any]], str \| None, gr.Dropdown, str]:
	"""Run at startup only when enabled in SETTINGS and a token is available."""
	if AUTO_RUN_ON_START and os.getenv("HF_TOKEN"):
	return run_smoke_test(progress=progress)
	return initial_view()


	def load_history(run_id: str \| None) -> tuple[str, list[list[Any]], str \| None]:
	if not run_id:
	run = smoke.load_latest_run()
	else:
	try:
	run = smoke.load_run(run_id)
	except ValueError as exc:
	raise gr.Error(str(exc)) from exc
	if run is None:
	raise gr.Error("That run could not be found in storage.")
	return _view(run)


	CSS = """
	#smoke-corner {
	position: fixed;
	right: 16px;
	bottom: 16px;
	width: 380px;
	max-width: calc(100vw - 32px);
	max-height: 70vh;
	overflow-y: auto;
	z-index: 1000;
	padding: 6px;
	border: 1px solid var(--border-color-primary);
	border-radius: 12px;
	background: var(--background-fill-primary);
	box-shadow: 0 12px 32px rgba(0, 0, 0, 0.18);
	}
	#smoke-corner .block {
	margin: 0;
	}
	#smoke-corner .label-wrap {
	padding: 4px 8px;
	}
	#smoke-corner .prose {
	font-size: 0.85rem;
	}
	#smoke-corner table {
	font-size: 0.78rem;
	}
	.status-row {
	display: flex;
	flex-wrap: wrap;
	gap: 8px;
	margin: 8px 0 16px;
	}
	.status-pill {
	display: inline-flex;
	align-items: center;
	gap: 6px;
	padding: 4px 10px;
	border-radius: 999px;
	border: 1px solid var(--border-color-primary);
	background: var(--background-fill-secondary);
	font-size: 0.8rem;
	line-height: 1.4;
	}
	.status-pill.ok {
	color: #0a7f3f;
	}
	.status-pill.warn {
	color: #b45309;
	}
	.status-dot {
	width: 8px;
	height: 8px;
	border-radius: 50%;
	background: currentColor;
	flex: 0 0 auto;
	}
	@media (max-width: 640px) {
	#smoke-corner {
	left: 10px;
	right: 10px;
	width: auto;
	max-height: 60vh;
	}
	}
	"""


	with gr.Blocks(title="StockAnalysis") as demo:
	gr.Markdown(HERO_MD)
	gr.HTML(_status_html())

	with gr.Row():
	with gr.Column(scale=1, min_width=280):
	ticker = gr.Textbox(
	label="Ticker",
	value="AAPL",
	placeholder="AAPL, MSFT, NVDA, BTC-USD",
	info="Symbol sent to the external project.",
	)
	horizon = gr.Dropdown(
	choices=HORIZONS,
	value="3 months",
	label="Horizon",
	)
	risk_profile = gr.Dropdown(
	choices=RISK_PROFILES,
	value="balanced",
	label="Risk profile",
	)
	question = gr.Textbox(
	label="Question (optional)",
	lines=3,
	placeholder="What should the analysis focus on?",
	)
	analyze_button = gr.Button("Analyze", variant="primary")
	gr.Markdown(DISCLAIMER_MD)

	with gr.Column(scale=2):
	result_status = gr.Markdown("Ready.")
	json_output = gr.JSON(
	label="JSON result",
	open=True,
	max_height=520,
	show_indices=False,
	)
	with gr.Accordion("Raw upstream response", open=False):
	raw_output = gr.Textbox(
	label="Raw response",
	lines=12,
	max_lines=20,
	interactive=False,
	buttons=["copy"],
	)

	with gr.Accordion("API / agents", open=False):
	gr.Markdown(_api_md())
	agents_text = _agents_text()
	if agents_text:
	gr.Code(
	agents_text,
	language="markdown",
	label="agents.md",
	lines=16,
	max_lines=30,
	wrap_lines=True,
	show_line_numbers=False,
	)
	if AGENTS_PATH.exists():
	gr.DownloadButton(
	"Download agents.md",
	value=str(AGENTS_PATH),
	size="sm",
	)
	else:
	gr.Markdown("The repository `agents.md` is not present.")

	with gr.Column(elem_id="smoke-corner"):
	with gr.Accordion("Inference smoke test", open=False):
	smoke_token_banner = gr.Markdown(_token_banner())
	smoke_run_button = gr.Button(
	"Run smoke test",
	size="sm",
	variant="secondary",
	)
	smoke_progress_note = gr.Markdown("Idle.")
	smoke_summary = gr.Markdown(_summary(None))
	smoke_table = gr.Dataframe(
	headers=SMOKE_TABLE_HEADERS,
	datatype=SMOKE_TABLE_TYPES,
	value=[],
	interactive=False,
	wrap=True,
	max_height=170,
	show_row_numbers=False,
	show_search=False,
	label="Models used (True = valid chat completion)",
	)
	with gr.Accordion("History & settings", open=False):
	smoke_history = gr.Dropdown(
	choices=[],
	value=None,
	label="Run history",
	info=f"Load a previous run from {RESULTS_DIR}/runs.",
	)
	smoke_download = gr.File(
	label="Download run JSON",
	interactive=False,
	)
	gr.Markdown(SETTINGS_MD)
	gr.Markdown(HOW_IT_WORKS)

	analyze_button.click(
	fn=analyze_stock,
	inputs=[ticker, horizon, risk_profile, question],
	outputs=[json_output, raw_output, result_status],
	api_name="analyze",
	api_description=(
	"Return the external stock-analysis project's response as JSON, "
	"the raw upstream body, and a short status string."
	),
	api_visibility="public",
	concurrency_limit=4,
	concurrency_id="analyze",
	)

	smoke_run_button.click(
	fn=run_smoke_test,
	inputs=[],
	outputs=[
	smoke_summary,
	smoke_table,
	smoke_download,
	smoke_history,
	smoke_token_banner,
	],
	api_visibility="private",
	concurrency_limit=1,
	concurrency_id="smoke",
	).success(
	lambda: "Run finished.",
	outputs=smoke_progress_note,
	api_visibility="private",
	)

	smoke_history.change(
	fn=load_history,
	inputs=[smoke_history],
	outputs=[smoke_summary, smoke_table, smoke_download],
	api_visibility="private",
	)

	demo.load(
	fn=maybe_auto_run,
	outputs=[
	smoke_summary,
	smoke_table,
	smoke_download,
	smoke_history,
	smoke_token_banner,
	],
	api_visibility="private",
	)


	if __name__ == "__main__":
	demo.launch(
	theme=gr.themes.Soft(),
	css=CSS,
	allowed_paths=[RESULTS_DIR],
	)