Spaces:

Darshan03
/

Test-app

Build error

Test-app / simluation_data.py

Upload 3 files

96e7d53 verified almost 2 years ago

6.95 kB

import json
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import pickle
# Function for Monte Carlo Simulation
def monte_carlo_simulation(portfolio_data, scenario_data, num_simulations=10000):
    """
    Performs a Monte Carlo simulation on a portfolio based on market scenarios.

    Args:
        portfolio_data (dict): Dictionary of portfolio data.
        scenario_data (dict): Dictionary of market scenario data.
        num_simulations (int, optional): The number of simulations. Defaults to 10000.

    Returns:
        dict: A dictionary containing simulation results for each scenario.
    """
    scenarios = scenario_data["market_scenarios"]
    results = {}

    for scenario_key, scenario_details in scenarios.items():
        scenario_name = scenario_details["name"]
        sector_impacts = scenario_details.get("sector_impact", {})
        results[scenario_name] = {
            "portfolio_values": [],
            "average_return": 0,
            "std_dev_return": 0,
            "percentiles": {},
        }

        for _ in range(num_simulations):
            portfolio_value = 0
            for asset_name, asset_details in portfolio_data["assets"].items():
                sector = asset_details["sector"]
                quantity = asset_details["quantity"]
                initial_price = asset_details["initial_price"]

                price_change_percentage = 0
                if sector in sector_impacts:
                    price_change_percentage = np.random.normal(
                        loc=sector_impacts[sector] / 100, scale=0.1
                    )

                # Calculate the new price
                new_price = initial_price * (1 + price_change_percentage)

                portfolio_value += new_price * quantity
            results[scenario_name]["portfolio_values"].append(portfolio_value)

        # Calculate Results
        portfolio_values = results[scenario_name]["portfolio_values"]
        initial_portfolio_value = sum(
            asset["quantity"] * asset["initial_price"] for asset in portfolio_data["assets"].values()
        )
        returns = [
            (value - initial_portfolio_value) / initial_portfolio_value
            for value in portfolio_values
        ]

        results[scenario_name]["average_return"] = np.mean(returns)
        results[scenario_name]["std_dev_return"] = np.std(returns)
        results[scenario_name]["percentiles"] = {
            5: np.percentile(returns, 5),
            25: np.percentile(returns, 25),
            50: np.percentile(returns, 50),
            75: np.percentile(returns, 75),
            95: np.percentile(returns, 95),
        }

    return results

if __name__ == "__main__":
    # Load input data
    with open("output_files/scenario.json") as f:
        scenario_data = json.load(f)

    with open("output_files/portfolio.json") as f:
        portfolio_data = json.load(f)


    # Load the dictionary from the local file
    def load_dataframes(filename="output_files/saved_dataframes.pkl"):
        try:
            with open(filename, 'rb') as file:
                saved_dataframes = pickle.load(file)
                print(f"DataFrames successfully loaded from {filename}.")
                return saved_dataframes
        except FileNotFoundError:
            print(f"File {filename} not found.")
            return None

    saved_dataframes = load_dataframes()

    # Placeholder for storing results
    scenario_results = {}

    # Process each scenario
    for scenario_name, scenario_details in scenario_data["market_scenarios"].items():
        impacted_sectors = scenario_details["sector_impact"]

        # Filter assets in the impacted sectors
        relevant_assets = [
            symbol
            for symbol, details in portfolio_data["assets"].items()
            if details["sector"] in impacted_sectors
        ]

        # Calculate magnitudes for the scenario
        sector_magnitudes = {}
        for symbol in relevant_assets:
            df = saved_dataframes[symbol]
            sector = portfolio_data["assets"][symbol]["sector"]

            # Calculate magnitude as the absolute difference between first and last Close price
            magnitude = abs(df["Close"].iloc[-2] - df["Close"].iloc[-1])

            # Aggregate by sector
            if sector not in sector_magnitudes:
                sector_magnitudes[sector] = 0
            sector_magnitudes[sector] += magnitude

        # Calculate aggregated magnitude for the scenario
        aggregated_magnitude = sum(sector_magnitudes.values())

        # Store results
        scenario_results[scenario_name] = {
            "individual_magnitudes": sector_magnitudes,
            "aggregated_magnitude": aggregated_magnitude,
        }

    # Display results
    for scenario_name, results in scenario_results.items():
        print(f"\nScenario: {scenario_name}")
        print("Individual Sector Magnitudes:")
        for sector, magnitude in results["individual_magnitudes"].items():
            print(f"  {sector}: {magnitude:.2f}")
        print(f"Aggregated Magnitude: {results['aggregated_magnitude']:.2f}")

    # Integrate calculated results into scenario data
    for scenario_id, results in scenario_results.items():
        # Update the sector impacts to include individual magnitudes
        scenario_data["market_scenarios"][scenario_id]["sector_impact"] = results["individual_magnitudes"]
        # Update aggregated magnitude
        scenario_data["market_scenarios"][scenario_id]["aggregated_magnitude"] = results["aggregated_magnitude"]

    # Save the updated scenario data to a local JSON file
    output_file_path = "output_files/updated_scenario_data.json"
    with open(output_file_path, "w") as file:
        json.dump(scenario_data, file, indent=4)

    print(f"Updated scenario data saved to '{output_file_path}' successfully!")

    # Run Monte Carlo simulation
    simulation_results = monte_carlo_simulation(portfolio_data, scenario_data)

    # Save simulation results to a local JSON file
    simulation_results_file = "output_files/simulation_results.json"
    with open(simulation_results_file, "w") as file:
        json.dump(simulation_results, file, indent=4)

    print(f"Simulation results saved to '{simulation_results_file}' successfully!")

    # Print simulation results
    for scenario_name, results in simulation_results.items():
        print(f"Scenario: {scenario_name}")
        print(f"  Average Return: {results['average_return']:.4f}")
        print(f"  Std Dev Return: {results['std_dev_return']:.4f}")
        print("  Return Percentiles:")
        for percentile, value in results["percentiles"].items():
            print(f"    {percentile}th: {value:.4f}")
        print("-" * 40)

	import json
	import numpy as np
	import pandas as pd
	import matplotlib.pyplot as plt
	import pickle
	# Function for Monte Carlo Simulation
	def monte_carlo_simulation(portfolio_data, scenario_data, num_simulations=10000):
	"""
	Performs a Monte Carlo simulation on a portfolio based on market scenarios.

	Args:
	portfolio_data (dict): Dictionary of portfolio data.
	scenario_data (dict): Dictionary of market scenario data.
	num_simulations (int, optional): The number of simulations. Defaults to 10000.

	Returns:
	dict: A dictionary containing simulation results for each scenario.
	"""
	scenarios = scenario_data["market_scenarios"]
	results = {}

	for scenario_key, scenario_details in scenarios.items():
	scenario_name = scenario_details["name"]
	sector_impacts = scenario_details.get("sector_impact", {})
	results[scenario_name] = {
	"portfolio_values": [],
	"average_return": 0,
	"std_dev_return": 0,
	"percentiles": {},
	}

	for _ in range(num_simulations):
	portfolio_value = 0
	for asset_name, asset_details in portfolio_data["assets"].items():
	sector = asset_details["sector"]
	quantity = asset_details["quantity"]
	initial_price = asset_details["initial_price"]

	price_change_percentage = 0
	if sector in sector_impacts:
	price_change_percentage = np.random.normal(
	loc=sector_impacts[sector] / 100, scale=0.1
	)

	# Calculate the new price
	new_price = initial_price * (1 + price_change_percentage)

	portfolio_value += new_price * quantity
	results[scenario_name]["portfolio_values"].append(portfolio_value)

	# Calculate Results
	portfolio_values = results[scenario_name]["portfolio_values"]
	initial_portfolio_value = sum(
	asset["quantity"] * asset["initial_price"] for asset in portfolio_data["assets"].values()
	)
	returns = [
	(value - initial_portfolio_value) / initial_portfolio_value
	for value in portfolio_values
	]

	results[scenario_name]["average_return"] = np.mean(returns)
	results[scenario_name]["std_dev_return"] = np.std(returns)
	results[scenario_name]["percentiles"] = {
	5: np.percentile(returns, 5),
	25: np.percentile(returns, 25),
	50: np.percentile(returns, 50),
	75: np.percentile(returns, 75),
	95: np.percentile(returns, 95),
	}

	return results

	if __name__ == "__main__":
	# Load input data
	with open("output_files/scenario.json") as f:
	scenario_data = json.load(f)

	with open("output_files/portfolio.json") as f:
	portfolio_data = json.load(f)


	# Load the dictionary from the local file
	def load_dataframes(filename="output_files/saved_dataframes.pkl"):
	try:
	with open(filename, 'rb') as file:
	saved_dataframes = pickle.load(file)
	print(f"DataFrames successfully loaded from {filename}.")
	return saved_dataframes
	except FileNotFoundError:
	print(f"File {filename} not found.")
	return None

	saved_dataframes = load_dataframes()

	# Placeholder for storing results
	scenario_results = {}

	# Process each scenario
	for scenario_name, scenario_details in scenario_data["market_scenarios"].items():
	impacted_sectors = scenario_details["sector_impact"]

	# Filter assets in the impacted sectors
	relevant_assets = [
	symbol
	for symbol, details in portfolio_data["assets"].items()
	if details["sector"] in impacted_sectors
	]

	# Calculate magnitudes for the scenario
	sector_magnitudes = {}
	for symbol in relevant_assets:
	df = saved_dataframes[symbol]
	sector = portfolio_data["assets"][symbol]["sector"]

	# Calculate magnitude as the absolute difference between first and last Close price
	magnitude = abs(df["Close"].iloc[-2] - df["Close"].iloc[-1])

	# Aggregate by sector
	if sector not in sector_magnitudes:
	sector_magnitudes[sector] = 0
	sector_magnitudes[sector] += magnitude

	# Calculate aggregated magnitude for the scenario
	aggregated_magnitude = sum(sector_magnitudes.values())

	# Store results
	scenario_results[scenario_name] = {
	"individual_magnitudes": sector_magnitudes,
	"aggregated_magnitude": aggregated_magnitude,
	}

	# Display results
	for scenario_name, results in scenario_results.items():
	print(f"\nScenario: {scenario_name}")
	print("Individual Sector Magnitudes:")
	for sector, magnitude in results["individual_magnitudes"].items():
	print(f" {sector}: {magnitude:.2f}")
	print(f"Aggregated Magnitude: {results['aggregated_magnitude']:.2f}")

	# Integrate calculated results into scenario data
	for scenario_id, results in scenario_results.items():
	# Update the sector impacts to include individual magnitudes
	scenario_data["market_scenarios"][scenario_id]["sector_impact"] = results["individual_magnitudes"]
	# Update aggregated magnitude
	scenario_data["market_scenarios"][scenario_id]["aggregated_magnitude"] = results["aggregated_magnitude"]

	# Save the updated scenario data to a local JSON file
	output_file_path = "output_files/updated_scenario_data.json"
	with open(output_file_path, "w") as file:
	json.dump(scenario_data, file, indent=4)

	print(f"Updated scenario data saved to '{output_file_path}' successfully!")

	# Run Monte Carlo simulation
	simulation_results = monte_carlo_simulation(portfolio_data, scenario_data)

	# Save simulation results to a local JSON file
	simulation_results_file = "output_files/simulation_results.json"
	with open(simulation_results_file, "w") as file:
	json.dump(simulation_results, file, indent=4)

	print(f"Simulation results saved to '{simulation_results_file}' successfully!")

	# Print simulation results
	for scenario_name, results in simulation_results.items():
	print(f"Scenario: {scenario_name}")
	print(f" Average Return: {results['average_return']:.4f}")
	print(f" Std Dev Return: {results['std_dev_return']:.4f}")
	print(" Return Percentiles:")
	for percentile, value in results["percentiles"].items():
	print(f" {percentile}th: {value:.4f}")
	print("-" * 40)