处理股票期权结算计算

import streamlit as st import pandas as pd from datetime import datetime, timedelta from decimal import Decimal, ROUND_HALF_UP

Set page title and icon

st.set_page_config(page_title="Futures Settlement Calculator", page_icon="") st.title("Stock Futures Settlement Price Calculation") st.markdown("Please enter the time range and data, supporting CSV upload or manual table input.")

--- 1. Time Setting Section ---

st.header("1. Set Sampling Time") col1, col2 = st.columns(2) with col1: start_time = st.time_input("Start Time", value=datetime.strptime("13:00", "%H:%M").time()) with col2: end_time = st.time_input("End Time", value=datetime.strptime("13:30", "%H:%M").time())

Calculate expected count

Convert time to today's full datetime object for calculation

today = datetime.now().date() dt_start = datetime.combine(today, start_time) dt_end = datetime.combine(today, end_time)

if dt_end <= dt_start: st.error("⚠️ The end time must be later than the start time") expected_count = 0 else: delta = dt_end - dt_start total_minutes = delta.total_seconds() / 60 expected_count = int(total_minutes * 12) st.info(f"⏱️ Total time: {int(total_minutes)} minutes | Expected data count: {expected_count} records")

--- 2. Data Input Section ---

st.header("2. Enter Data") input_method = st.radio("Select input method:", ["Upload CSV file", "Manual input/paste"])

df_data = None if input_method == "Upload CSV file": uploaded_file = st.file_uploader("Please upload CSV (no header, A column price, B column count)", type=["csv"]) if uploaded_file is not None: try: # Read CSV, assuming no header df_data = pd.read_csv(uploaded_file, header=None, names=["Price", "Count"]) except Exception as e: st.error(f"Read error: {e}") else: # Manual input st.caption("Please enter directly in the table below, or copy and paste from Excel (click the table's upper right corner to zoom in)") # Create an empty DataFrame for user editing # num_rows="dynamic" allows users to add rows df_input = pd.DataFrame([{"Price": 150.0, "Count": 1}]) df_data = st.data_editor(df_input, num_rows="dynamic", use_container_width=True)

--- 3. Core Calculation ---

if st.button("Start calculating settlement price", type="primary"): if df_data is None or df_data.empty: st.warning("Please enter or upload data first.") elif expected_count == 0: st.warning("The time setting is incorrect.") else: try: total_weighted_sum = Decimal('0.0') total_count = 0

        # Iterate through data (convert to Decimal to ensure precision)
        # Using iterrows is slower, but convenient for type conversion for several hundred data rows
        for index, row in df_data.iterrows():
            try:
                p = Decimal(str(row['Price']).strip())
                c = int(row['Count'])
                total_weighted_sum += p * c
                total_count += c
            except:
                continue # skip invalid row
        
        if total_count == 0:
            st.error("No valid data rows (please check the format).")
        else:
            st.divider()
            
            # Verify quantity
            diff = expected_count - total_count
            if diff != 0:
                status = "less than" if diff > 0 else "more than"
                st.warning(f"⚠️ Warning: Actual sample {total_count} records, {status} expected ({abs(diff)} records).")
            else:
                st.success(f"✅ Sample quantity is correct ({total_count} records)")
            
            # Calculate
            average =