Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# Import main API functions
from api.interface import plot, stop_server, get_server_status
from pycharting.api.interface import plot, stop_server, get_server_status

__all__ = ["plot", "stop_server", "get_server_status", "__version__"]
66 changes: 60 additions & 6 deletions src/pycharting/api/interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import webbrowser
import time
import logging
import socket
from typing import Optional, Dict, Any, Union
import numpy as np
import pandas as pd
Expand All @@ -28,6 +29,38 @@

logger = logging.getLogger(__name__)


def _get_local_ip() -> str:
"""Get the local machine's IP address that can be reached externally.

Uses a UDP socket trick - connects to an external address without sending data,
then reads which local IP was used. Falls back to hostname resolution or localhost.
"""
try:
# Create UDP socket (doesn't actually send anything)
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0.1)
# Connect to a public IP (Google DNS) - this doesn't send data,
# just determines which local interface would be used
s.connect(("8.8.8.8", 80))
local_ip = s.getsockname()[0]
s.close()
return local_ip
except Exception:
pass

# Fallback: try hostname resolution
try:
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
if not local_ip.startswith("127."):
return local_ip
except Exception:
pass

# Last resort
return "127.0.0.1"

# Global server instance
_active_server: Optional[ChartServer] = None

Expand All @@ -41,10 +74,12 @@ def plot(
overlays: Optional[Dict[str, Union[np.ndarray, pd.Series, list]]] = None,
subplots: Optional[Dict[str, Union[np.ndarray, pd.Series, list]]] = None,
session_id: str = "default",
host: str = "127.0.0.1",
port: Optional[int] = None,
open_browser: bool = True,
server_timeout: float = 2.0,
block: bool = True,
external_host: Optional[str] = None,
) -> Dict[str, Any]:
"""
Generate and display an interactive OHLC (Open-High-Low-Close) or Line chart.
Expand All @@ -66,16 +101,23 @@ def plot(
low (Optional[Union[np.ndarray, pd.Series, list]]): Lowest prices during the interval.
close (Optional[Union[np.ndarray, pd.Series, list]]): Closing prices. If only `close` is provided (without open/high/low),
a line chart will be rendered instead of candlesticks.
overlays (Optional[Dict[str, Union[np.ndarray, pd.Series, list]]]): A dictionary of additional series to plot *over* the main price chart.
Keys are labels (e.g., "SMA 50"), values are data arrays. Useful for Moving Averages, Bollinger Bands, etc.
overlays (Optional[Dict[str, Union[np.ndarray, pd.Series, list, dict]]]): A dictionary of additional series to plot *over* the main price chart.
Keys are labels. Values can be:
- Simple: data array (renders as line) - e.g., `{"SMA 50": sma_array}`
- Styled: dict with `data`, `style`, `color`, `size` - e.g., `{"Markers": {"data": arr, "style": "marker", "color": "#00C853"}}`
Supported styles: "line" (default), "marker" (points), "dashed" (dashed line).
subplots (Optional[Dict[str, Union[np.ndarray, pd.Series, list]]]): A dictionary of series to plot in separate panels *below* the main chart.
Keys are labels (e.g., "RSI", "Volume"), values are data arrays.
session_id (str): A unique identifier for this dataset. Use different IDs to keep multiple charts active simultaneously.
Defaults to "default".
host (str): Host address to bind the server to. Use "0.0.0.0" to allow external connections (e.g., from Docker host).
Defaults to "127.0.0.1" (localhost only).
port (Optional[int]): Specific port to run the server on. If `None` (default), a free port is automatically found.
open_browser (bool): If `True` (default), automatically launches the system's default web browser to view the chart.
server_timeout (float): Maximum time (in seconds) to wait for the server to start before proceeding. Defaults to 2.0.
block (bool): If `True` (default), blocks execution until the browser page is closed. Useful in Jupyter notebooks.
external_host (Optional[str]): Override hostname/IP shown in URLs. When binding to "0.0.0.0", the local IP
is auto-detected. Use this parameter to override with a specific hostname or IP if needed.

Returns:
Dict[str, Any]: A dictionary containing execution details:
Expand Down Expand Up @@ -160,9 +202,9 @@ def plot(
if _active_server is None or not _active_server.is_running:
logger.info("Starting ChartServer...")
_active_server = ChartServer(
host="127.0.0.1",
host=host,
port=port,
auto_shutdown_timeout=3.0 # 3 seconds after disconnect
auto_shutdown_timeout=60.0 # 60 seconds after disconnect (handles browser throttling)
)
server_info = _active_server.start_server()

Expand All @@ -177,7 +219,19 @@ def plot(
# Construct chart URL with session ID and timestamp to bust cache
# Use viewport demo which pulls data from the API for the given session
ts = int(time.time())
chart_url = f"{server_info['url']}/static/viewport-demo.html?session={session_id}&v={ts}"

# Determine display host for URLs
# Priority: external_host > auto-detect (if 0.0.0.0) > server bind host
if external_host:
display_host = external_host
elif server_info['host'] == "0.0.0.0":
# Auto-detect local IP when binding to all interfaces
display_host = _get_local_ip()
else:
display_host = server_info['host']

display_url = f"http://{display_host}:{server_info['port']}"
chart_url = f"{display_url}/static/viewport-demo.html?session={session_id}&v={ts}"

# Open browser if requested
if open_browser:
Expand All @@ -191,7 +245,7 @@ def plot(
result = {
"status": "success",
"url": chart_url,
"server_url": server_info['url'],
"server_url": display_url,
"session_id": session_id,
"data_points": data_manager.length,
"server_running": _active_server.is_running if _active_server else False,
Expand Down
17 changes: 15 additions & 2 deletions src/pycharting/api/routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,15 +26,28 @@
_data_managers: Dict[str, Any] = {}


class OverlayData(BaseModel):
"""Overlay data with style configuration."""
data: list
style: str = "line" # "line", "marker", "dashed"
color: Optional[str] = None
size: Optional[int] = None


class DataResponse(BaseModel):
"""Response model for data endpoint."""
index: list
open: Optional[list] = None
high: Optional[list] = None
low: Optional[list] = None
close: Optional[list] = None
overlays: Dict[str, list] = Field(default_factory=dict)
subplots: Dict[str, list] = Field(default_factory=dict)
# Overlays now include style metadata: {data, style, color, size}
overlays: Dict[str, OverlayData] = Field(default_factory=dict)
# Subplots can be:
# - Simple: {"RSI": [values...]}
# - Grouped: {"Stochastic": {"%K": [values...], "%D": [values...]}}
# - Histogram: {"Filter": {"_type": "histogram", "Buy": [values...], "Sell": [values...]}}
subplots: Dict[str, Any] = Field(default_factory=dict)
start_index: int
end_index: int
total_length: int
Expand Down
73 changes: 61 additions & 12 deletions src/pycharting/data/ingestion.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,13 @@
The `DataManager` class is the core component here, acting as the optimized data store for a chart session.
"""

from typing import Optional, Union, List, Dict, Any
from typing import Optional, Union, Dict, Any
import pandas as pd
import numpy as np

SubplotSeries = Union[pd.Series, np.ndarray, list]
SubplotsInput = Dict[str, Union[SubplotSeries, Dict[str, SubplotSeries]]]


class DataValidationError(Exception):
"""Exception raised when input data fails validation checks."""
Expand All @@ -27,7 +30,7 @@ def validate_input(
low: Optional[Union[pd.Series, np.ndarray]] = None,
close: Optional[Union[pd.Series, np.ndarray]] = None,
overlays: Optional[Dict[str, Union[pd.Series, np.ndarray]]] = None,
subplots: Optional[Dict[str, Union[pd.Series, np.ndarray]]] = None,
subplots: Optional[SubplotsInput] = None,
) -> Dict[str, Any]:
"""
Validate and normalize input data for OHLC charting.
Expand Down Expand Up @@ -159,16 +162,45 @@ def to_array(data: Optional[Union[pd.Series, np.ndarray, list]], name: str) -> O
}

# Validate and convert overlays
# Supports two formats:
# - {"SMA 20": array} # Simple line overlay
# - {"Markers": {"data": array, "style": "marker", ...}} # Styled overlay
if overlays:
for name, data in overlays.items():
arr = to_array(data, f"Overlay '{name}'")
result["overlays"][name] = arr
if isinstance(data, dict) and "data" in data:
# Styled overlay format
arr = to_array(data["data"], f"Overlay '{name}'")
result["overlays"][name] = {
"data": arr,
"style": data.get("style", "line"), # "line", "marker", "dashed"
"color": data.get("color"), # Optional custom color
"size": data.get("size"), # Optional marker size
}
else:
# Simple format - just the array (backward compatible)
arr = to_array(data, f"Overlay '{name}'")
result["overlays"][name] = {"data": arr, "style": "line"}

# Validate and convert subplots
# Validate and convert subplots.
# Supports three formats:
# - {"RSI": array} # Simple line subplot
# - {"Stochastic": {"%K": array, "%D": array}} # Grouped line subplot
# - {"Filter": {"_type": "histogram", "Buy": array, "Sell": array}} # Histogram subplot
if subplots:
for name, data in subplots.items():
arr = to_array(data, f"Subplot '{name}'")
result["subplots"][name] = arr
if isinstance(data, dict):
nested: dict[str, Any] = {}
for series_name, series_data in data.items():
# Preserve _type metadata key (not an array)
if series_name == "_type":
nested["_type"] = series_data
else:
arr = to_array(series_data, f"Subplot '{name}.{series_name}'")
nested[series_name] = arr
result["subplots"][name] = nested
else:
arr = to_array(data, f"Subplot '{name}'")
result["subplots"][name] = arr

return result

Expand All @@ -186,7 +218,7 @@ def __init__(
low: Optional[Union[pd.Series, np.ndarray]] = None,
close: Optional[Union[pd.Series, np.ndarray]] = None,
overlays: Optional[Dict[str, Union[pd.Series, np.ndarray]]] = None,
subplots: Optional[Dict[str, Union[pd.Series, np.ndarray]]] = None,
subplots: Optional[SubplotsInput] = None,
):
# Validate input and get normalized arrays
validated = validate_input(index, open, high, low, close, overlays, subplots)
Expand Down Expand Up @@ -215,7 +247,7 @@ def close(self) -> Optional[np.ndarray]: return self._close
@property
def overlays(self) -> Dict[str, np.ndarray]: return self._overlays
@property
def subplots(self) -> Dict[str, np.ndarray]: return self._subplots
def subplots(self) -> Dict[str, Any]: return self._subplots
@property
def length(self) -> int: return self._length
def __len__(self) -> int: return self._length
Expand Down Expand Up @@ -268,10 +300,27 @@ def slice_opt(arr):
"subplots": {},
}

for name, data in self._overlays.items():
result["overlays"][name] = data[start_index:end_index].tolist()
for name, overlay_info in self._overlays.items():
# Overlay is now a dict with "data" array and optional style metadata
data_arr = overlay_info["data"]
result["overlays"][name] = {
"data": data_arr[start_index:end_index].tolist(),
"style": overlay_info.get("style", "line"),
"color": overlay_info.get("color"),
"size": overlay_info.get("size"),
}

for name, data in self._subplots.items():
result["subplots"][name] = data[start_index:end_index].tolist()
if isinstance(data, dict):
nested: dict[str, Any] = {}
for series_name, series_data in data.items():
# Preserve _type metadata (not an array, just pass through)
if series_name == "_type":
nested["_type"] = series_data
else:
nested[series_name] = series_data[start_index:end_index].tolist()
result["subplots"][name] = nested
else:
result["subplots"][name] = data[start_index:end_index].tolist()

return result
Loading