Market Candles example.

[mifth]

Hi, I would like to contribute an example but I'm lazy to do forking etc. I'm just pasting my code here in hope that one hero will be able to polish it and check in.

import pandas as pd
import numpy as np
import plotly.graph_objects as go
import panel as pn

pn.extension('plotly')


class SessionPanel:
    def __init__(self):
        self.n = 50
        self.df: pd.DataFrame = self._make_df()
        self.fig: go.Figure = self._make_fig()
        self.plotly_pane: pn.pane.Plotly = pn.pane.Plotly(self.fig, sizing_mode='stretch_both')
        # self.plotly_pane.object = self.fig  # Example of setting the object directly

    def __del__(self):
        print("SessionPanel instance is being deleted!!!")

    def _make_df(self) -> pd.DataFrame:
        dates = pd.date_range("2023-01-01", periods=self.n, freq='min')
        price = np.cumsum(np.random.randn(self.n)) + 100
        df = pd.DataFrame(dict(
            Date=dates,
            Open=price + np.random.uniform(-1, 1, self.n),
        ))
        df['Close'] = df['Open'] + np.random.uniform(-2, 2, self.n)
        df['High'] = np.maximum(df['Open'], df['Close']) + np.random.uniform(0, 1, self.n)
        df['Low'] = np.minimum(df['Open'], df['Close']) - np.random.uniform(0, 1, self.n)
        return df

    def _make_fig(self) -> go.Figure:
        fig = go.Figure(
            go.Candlestick(
                x=self.df['Date'],
                open=self.df['Open'],
                high=self.df['High'],
                low=self.df['Low'],
                close=self.df['Close']
            )
        )
        fig.update_layout(
            title="Live Candlestick Chart",
            xaxis_rangeslider_visible=False
        )
        return fig

    def update(self):
        # Create one new row of data
        new_date = self.df['Date'].iloc[-1] + pd.Timedelta(minutes=1)
        new_open = self.df['Close'].iloc[-1] + np.random.uniform(-1, 1)
        new_close = new_open + np.random.uniform(-2, 2)
        new_high = max(new_open, new_close) + np.random.uniform(0, 1)
        new_low = min(new_open, new_close) - np.random.uniform(0, 1)

        new_row = pd.DataFrame(dict(
            Date=[new_date],
            Open=[new_open],
            High=[new_high],
            Low=[new_low],
            Close=[new_close]
        ))

        # Slide the window
        self.df = pd.concat([self.df.iloc[1:], new_row], ignore_index=True)

        # Patch only the changed data in the dict
        self.plotly_pane.object['data'][0]['x'] = self.df['Date']
        self.plotly_pane.object['data'][0]['open'] = self.df['Open']
        self.plotly_pane.object['data'][0]['high'] = self.df['High']
        self.plotly_pane.object['data'][0]['low'] = self.df['Low']
        self.plotly_pane.object['data'][0]['close'] = self.df['Close']

class DashboardPanel(pn.Column):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Create session-unique data and plot
        self.session = SessionPanel()

        # Widgets
        self.button = pn.widgets.Button(name='Click Me', button_type='primary')
        self.select_opt = pn.widgets.Select(name='Market', options=['BTC/USD', 'ETH/USD', 'SPY'])
        self.text = pn.widgets.TextInput(name='Notes', placeholder='Enter notes...')

        self.button.on_click(self.say_hello)

        self.tab1 = pn.Column(
            "## Controls",
            self.button,
            self.select_opt,
        )

        self.tab2 = pn.Column(
            "## Notes",
            self.text
        )

        self.tabs = pn.Tabs(
            ("Tab 1", self.tab1),
            ("Tab 2", self.tab2)
        )

        self.dashboard = pn.Row(
            pn.Column("# Left Panel", self.tabs, width=300, sizing_mode='stretch_height'),
            pn.Column("# Right Panel", self.session.plotly_pane, sizing_mode='stretch_both'),
            sizing_mode='stretch_both'
        )

        # Add the dashboard layout to this Column
        self.append(self.dashboard)

        # Add periodic callback for updates (session-specific)
        pn.state.add_periodic_callback(self.session.update, period=1000)

        # Add session cleanup callback
        pn.state.on_session_destroyed(self.on_session_destroyed)

    def say_hello(self, event):
        print(f"Button clicked! Market: {self.select_opt.value}, Note: {self.text.value}")

    def on_session_destroyed(self, session_context):
        print("Session ended!")


# Show dashboard based on session
def app():
    main_panel: pn.Column = pn.Column()
    main_panel.append(DashboardPanel())
    return main_panel

app().servable()

[MarcSkovMadsen]

Here is a best practice refactor from my AI :-)

live-stock-dashboard.mp4

Todo

• Add nice background color to sidebar
• Data is no longer sliding in a window. Its just replaced. Fix.

(I don't plan to do this)

Code

"""
Live Stock Dashboard using Panel Material UI

This application demonstrates a live stock candlestick chart with Material UI components
following Panel Material UI best practices.

Example:
    Run the application:
    
    $ python stock_app.py
    $ panel serve stock_app.py --port 5007 --dev --show
"""
from typing import Optional
import pandas as pd
import numpy as np
import plotly.graph_objects as go
import panel as pn
import panel_material_ui as pmui
import param

pn.extension('plotly', notifications=True)


# DO organize functions to extract data separately as your app grows
def extract_stock_data(n_points: int = 50, market: str = 'BTC/USD') -> pd.DataFrame:
    """
    Extract initial stock data for the specified market.
    
    Args:
        n_points: Number of data points to generate
        market: Market symbol for data generation
        
    Returns:
        DataFrame with stock data columns: Date, Open, Close, High, Low
    """
    dates = pd.date_range("2023-01-01", periods=n_points, freq='min')
    # Use market symbol to create different price patterns
    base_price = 100 if 'BTC' in market else 50 if 'ETH' in market else 200
    price = np.cumsum(np.random.randn(n_points)) + base_price
    
    df = pd.DataFrame({
        'Date': dates,
        'Open': price + np.random.uniform(-1, 1, n_points),
    })
    df['Close'] = df['Open'] + np.random.uniform(-2, 2, n_points)
    df['High'] = np.maximum(df['Open'], df['Close']) + np.random.uniform(0, 1, n_points)
    df['Low'] = np.minimum(df['Open'], df['Close']) - np.random.uniform(0, 1, n_points)
    return df


# DO organize functions to transform data separately as your app grows
def transform_stock_data(df: pd.DataFrame, window_size: int = 50) -> pd.DataFrame:
    """
    Transform stock data by adding new data points and maintaining window size.
    
    Implements a true moving window where old data points are removed as new ones
    are added, maintaining exactly the specified window size.
    
    Args:
        df: Existing stock data
        window_size: Number of data points to maintain
        
    Returns:
        Updated DataFrame with new data point and sliding window
    """
    # Create one new row of data
    new_date = df['Date'].iloc[-1] + pd.Timedelta(minutes=1)
    new_open = df['Close'].iloc[-1] + np.random.uniform(-1, 1)
    new_close = new_open + np.random.uniform(-2, 2)
    new_high = max(new_open, new_close) + np.random.uniform(0, 1)
    new_low = min(new_open, new_close) - np.random.uniform(0, 1)

    new_row = pd.DataFrame({
        'Date': [new_date],
        'Open': [new_open],
        'High': [new_high],
        'Low': [new_low],
        'Close': [new_close]
    })

    # Add new row to the dataframe
    updated_df = pd.concat([df, new_row], ignore_index=True)
    
    # Implement sliding window - keep only the last window_size points
    if len(updated_df) > window_size:
        updated_df = updated_df.tail(window_size).reset_index(drop=True)
    
    return updated_df


# DO organize functions to create plots separately as your app grows
# Note: No caching used since data is live and constantly changing
def create_candlestick_plot(df: pd.DataFrame, title: str = "Live Candlestick Chart", dark_theme: bool = False) -> go.Figure:
    """
    Create Plotly candlestick figure from stock data.
    
    Args:
        df: Stock data DataFrame
        title: Chart title
        dark_theme: Whether to use dark theme for the plot
        
    Returns:
        Plotly Figure object
    """
    print("theme", dark_theme)
    fig = go.Figure(
        go.Candlestick(
            x=df['Date'],
            open=df['Open'],
            high=df['High'],
            low=df['Low'],
            close=df['Close']
        )
    )
    
    # Choose template based on theme
    template = "plotly_dark" if dark_theme else "plotly_white"
    
    fig.update_layout(
        title=title,
        xaxis_rangeslider_visible=False,
        template=template,
        margin=dict(l=40, r=40, t=60, b=40),  # Add margins for better spacing
        height=500,  # Set reasonable height
    )
    return fig


class StockDashboard(pn.viewable.Viewer):
    """
    Live Stock Dashboard with Material UI components.
    
    A comprehensive dashboard that displays live updating candlestick charts with controls
    for market selection and note-taking functionality. Built following Panel Material UI
    best practices with parameter-driven reactivity.
    
    References:
        - https://panel-material-ui.holoviz.org/explanation/best_practices.html
        - https://panel.holoviz.org/tutorials/intermediate/index.html
    
    Example:
        >>> dashboard = StockDashboard()
        >>> dashboard  # Display in notebook
        
        >>> # Or create a servable app
        >>> StockDashboard.create_app().servable()
    """
    
    # DO define parameters to hold state and drive the reactivity
    market = param.Selector(
        default='BTC/USD', 
        objects=['BTC/USD', 'ETH/USD', 'SPY'],
        doc="Selected market to display"
    )
    notes = param.String(
        default='', 
        doc="User notes about the market analysis"
    )
    update_enabled = param.Boolean(
        default=True,
        doc="Whether to enable live chart updates"
    )
    dark_theme = param.Boolean(
        default=False,
        doc="Whether to use dark theme for the dashboard",
        allow_refs=True
    )
    
    def __init__(self, **params):
        super().__init__(**params)
        
        # Initialize stock data
        market_str = str(getattr(self, 'market', 'BTC/USD'))
        self.stock_data = extract_stock_data(market=market_str)
        
        # DO use sizing_mode="stretch_width" for components unless "fixed" or other sizing_mode is specifically needed
        # Create widgets and layouts
        with pn.config.set(sizing_mode="stretch_width"):
            self._create_widgets()
            
            # DO Collect input widgets into horizontal, columnar layout unless other layout is specifically needed
            self._inputs = pmui.Column(
                pmui.Typography("📈 Trading Dashboard", variant="h6", sx={"mb": 2, "color": "primary.main"}),
                pmui.Typography("Monitor live market data with analysis tools.", variant="body2", sx={"mb": 3, "color": "text.secondary"}),
                self._market_input,
                self._update_toggle,
                self._notes_input,
                self._action_button,
                sx={"overflowY": "auto", "overflowX": "hidden", "padding": "10px"}  # Ensure proper scrolling behavior
            )
            # DO collect output components into some layout like Column, Row, FlexBox or Grid depending on use case
            self._outputs = pmui.Column(
                pmui.Container(
                    pmui.Typography("Live Market Data", variant="h5", margin=(0, 20)),
                    pn.pane.Plotly(
                        self.chart_pane,
                        sizing_mode='stretch_width',
                        height=500,  # Fixed height to prevent overflow
                        max_width=1000,  # Maximum width for better centering
                        max_height=500,  # Maximum height to prevent overflow
                        margin=(30, 30),  # Increased margin for better spacing
                        styles={"display": "block", "margin": "0 auto"}  # Center the plot
                    ),
                    sx={
                        "display": "flex",
                        "justifyContent": "center",
                        "alignItems": "center",
                        "padding": "20px",
                        "maxWidth": "100%",
                        "width": "100%",
                        "margin": "0 auto"
                    }
                ),
                sizing_mode="stretch_both",
                margin=(20, 20)  # Add margin around the entire output area
            )
            # Create main layout
            self._panel = pmui.Row(
                pmui.Paper(self._inputs, elevation=2, sx={"padding": "20px"}),
                pmui.Paper(self._outputs, elevation=1, sx={"padding": "20px", "margin": "0 0 0 10px"}),
                sizing_mode="stretch_both"
            )
        
        # Set up callbacks and periodic updates
        self._setup_callbacks()
    
    def _create_widgets(self):
        """Create all UI widgets from parameters following best practices."""
        # DO create widgets using `.from_param` method
        self._market_input = pmui.Select.from_param(
            self.param.market,
            label="Market Symbol",
        )
        
        self._update_toggle = pmui.Switch.from_param(
            self.param.update_enabled,
            label="Live Updates",
        )
        
        self._notes_input = pmui.TextAreaInput.from_param(
            self.param.notes,
            label="Analysis Notes",
            placeholder="Enter your market analysis notes...",
            max_rows=3,  # Reduce rows to save space
        )
        
        # Action button (not parameter-driven)
        self._action_button = pmui.Button(
            label="Save Analysis",
            color="primary",
            icon="save",
        )
    
    # DO prefer .depends over .bind over .rx for reactivity methods on Parameterized classes as it can be typed and documented
    # Note: No caching used here since data is live and constantly changing
    @param.depends("market", "dark_theme")
    def chart_pane(self):
        """
        Returns the live candlestick chart for the selected market.
        
        This method is reactive and will update when the market or dark_theme parameter changes.
        No caching is used since the data is live and constantly updating.
        
        Returns:
            pn.pane.Plotly: Interactive candlestick chart
        """
        # Update stock data for new market if needed
        if not hasattr(self, 'stock_data') or len(self.stock_data) == 0:
            market_str = str(getattr(self, 'market', 'BTC/USD'))
            self.stock_data = extract_stock_data(market=market_str)
        
        # Create chart with current data and theme
        market_str = self.market
        dark_theme = self.dark_theme
        fig = create_candlestick_plot(
            self.stock_data, 
            title=f"Live {market_str} Chart",
            dark_theme=dark_theme
        )
        
        return fig
    
    def _setup_callbacks(self):
        """Set up event handlers and periodic updates."""
        # Set up action button callback
        self._action_button.on_click(self._handle_save_action)
        
        # DO use `watch=True` or `.watch` for triggering side effect like saving file or sending email.
        self.param.watch(self._on_market_change, 'market')
        
        # Set up periodic callback for live updates
        if pn.state.served:
            self._update_callback_id = pn.state.add_periodic_callback(
                self._update_chart_data, 
                period=2000,  # Update every 2 seconds
                start=True
            )
            
            # Add session cleanup
            pn.state.on_session_destroyed(self._cleanup_session)
    
    def _on_market_change(self, event):
        """Handle market selection changes."""
        # Reset stock data for new market
        market_str = str(getattr(self, 'market', 'BTC/USD'))
        self.stock_data = extract_stock_data(market=market_str)
    
    def _handle_save_action(self, event):
        """Handle save action button click."""
        market_str = str(getattr(self, 'market', 'BTC/USD'))
        message = f"Analysis saved for {market_str}!"
        notes_value = getattr(self, 'notes', '')
        if notes_value and str(notes_value).strip():
            note_preview = str(notes_value)[:50]
            note_preview += '...' if len(str(notes_value)) > 50 else ''
            message += f" Notes: {note_preview}"
        
        # Show success notification if available
        if pn.state.notifications:
            pn.state.notifications.success(
                message,
                duration=3000  # Show for 3 seconds
            )
    
    def _update_chart_data(self):
        """
        Update chart data with new data points if updates are enabled.
        
        DO feel free to trigger non-UI, side-effects like external computations or sending emails 
        via `.watch` or `watch=True` functions and methods.
        """
        update_enabled = getattr(self, 'update_enabled', True)
        if not update_enabled:
            return
            
        try:
            # Transform stock data by adding new point
            self.stock_data = transform_stock_data(self.stock_data)
            
            # Trigger chart update by changing market parameter
            # This will cause the chart_pane method to re-execute
            self.param.trigger('market')
            
        except Exception as e:
            error_message = f"Error updating chart: {e}"
            # Show error notification if available
            if pn.state.notifications:
                pn.state.notifications.error(
                    error_message,
                    duration=5000  # Show errors longer (5 seconds)
                )
    
    def _cleanup_session(self, session_context):
        """Clean up when session is destroyed."""
        print("[Stock Dashboard] Session ended, cleaning up...")
        # Panel handles periodic callback cleanup automatically
    
    # DO provide a method for displaying the component in a notebook setting
    def __panel__(self):
        """Return the main layout for notebook display."""
        return self._panel
    
    # DO provide a method to create a .servable app
    @classmethod
    def create_app(cls, **params):
        """
        Create the Panel app with Page layout for serving.
        
        Args:
            **params: Additional parameters to pass to the dashboard instance
            
        Returns:
            pmui.Page: Complete app ready to serve
        """
        instance = cls(**params)
        
        # DO use the `Page` to layout the served app unless otherwise specified
        page = pmui.Page(
            # DO provide a title for the app
            title="Live Stock Dashboard",
            # DO provide optional image, optional app description, optional navigation menu, input widgets, optional documentation and optional links in the sidebar
            # DO provide as list of components or a list of single horizontal layout like Column as the sidebar by default is 300 px wide
            sidebar=list(instance._inputs),
            # DO provide a list of layouts and output components in the main area of the app.
            main=list(instance._outputs),
            # Pass dark theme parameter to Page
            dark_theme=instance.dark_theme,
        )
        instance.dark_theme = page.param.dark_theme
        return page


# DO provide a method to serve the app with `python`
if __name__ == "__main__":
    # DO run with `python path_to_this_file.py`
    # DO prefer to serve with `panel serve path_to_this_file.py --dev --show` instead of `python path_to_this_file.py`.
    StockDashboard.create_app().show(port=5007, autoreload=True, open=True)
# DO provide a method to serve the app with `panel serve`
elif pn.state.served:
    # DO run with `panel serve path_to_this_file.py --port 5007 --dev` add `--show` to open the app in a browser
    StockDashboard.create_app().servable()  # DO mark the element(s) to serve with .servable()

[mifth]

Oooooooooooo!!! It looks awesome!

[mifth]

Hey @MarcSkovMadsen , thank you so much for your improvements. I tried your example but got an empty page. I run the script from python.

Log:
theme False
Launching server at http://localhost:5007

[mifth]

I also noticed that I had to install the panel_material_ui lib with pip. My script didn't use it. But that's not a problem at all.