Spaces:
Configuration error
Configuration error
| """ | |
| BI Storyteller CLI Interface | |
| Command-line interface for marketing analysis automation | |
| Standard Library Only - No Network Dependencies | |
| """ | |
| import json | |
| import os | |
| from main import BIStoryteller | |
| class BIStoryteller_CLI: | |
| """Command-line interface for BI Storyteller""" | |
| def __init__(self): | |
| self.bi = BIStoryteller() | |
| self.current_step = 1 | |
| def print_header(self): | |
| """Print application header""" | |
| print("\n" + "="*60) | |
| print("π BI STORYTELLER - MARKETING ANALYSIS PLATFORM") | |
| print("="*60) | |
| print("π Complete workflow for marketing data analysis") | |
| print("π§ Standard Library Only - No External Dependencies") | |
| print("="*60) | |
| def print_menu(self): | |
| """Print main menu""" | |
| print(f"\nπ MAIN MENU (Current Step: {self.current_step}/12)") | |
| print("-" * 40) | |
| print("1. π Set API Key (Optional)") | |
| print("2. π Extract Variables") | |
| print("3. π Generate Questionnaire") | |
| print("4. π’ Generate Sample Data") | |
| print("5. π§Ή Clean Data") | |
| print("6. π Perform EDA") | |
| print("7. π€ Train Predictive Model") | |
| print("8. π Analyze Trends") | |
| print("9. π Analyze Sentiment") | |
| print("10. π§ͺ Run A/B Test") | |
| print("11. π¬ Chat with Data") | |
| print("12. π€ Export Results") | |
| print("-" * 40) | |
| print("13. π₯ Import Previous Analysis") | |
| print("14. π Export Data as CSV") | |
| print("15. β Exit") | |
| print("-" * 40) | |
| def get_user_input(self, prompt, input_type="string"): | |
| """Get user input with validation""" | |
| while True: | |
| try: | |
| user_input = input(f"\n{prompt}: ").strip() | |
| if input_type == "int": | |
| return int(user_input) | |
| elif input_type == "float": | |
| return float(user_input) | |
| else: | |
| return user_input | |
| except ValueError: | |
| print(f"β Please enter a valid {input_type}") | |
| except KeyboardInterrupt: | |
| print("\nπ Goodbye!") | |
| exit(0) | |
| def print_results(self, title, results, success_key="success"): | |
| """Print formatted results""" | |
| print(f"\n{title}") | |
| print("-" * len(title)) | |
| if results.get(success_key): | |
| if "results" in results: | |
| self.print_dict(results["results"], indent=0) | |
| else: | |
| self.print_dict(results, indent=0) | |
| else: | |
| print(f"β Error: {results.get('error', 'Unknown error')}") | |
| def print_dict(self, data, indent=0): | |
| """Print dictionary in a formatted way""" | |
| spaces = " " * indent | |
| for key, value in data.items(): | |
| if isinstance(value, dict): | |
| print(f"{spaces}{key}:") | |
| self.print_dict(value, indent + 1) | |
| elif isinstance(value, list): | |
| print(f"{spaces}{key}: [{len(value)} items]") | |
| if value and len(value) <= 5: | |
| for item in value: | |
| print(f"{spaces} β’ {item}") | |
| else: | |
| print(f"{spaces}{key}: {value}") | |
| def module_1_api_key(self): | |
| """Module 1: Set API Key""" | |
| print("\nπ MODULE 1: API KEY SETUP") | |
| print("=" * 30) | |
| print("Enter your Groq API key for AI-powered analysis.") | |
| print("Leave empty to use offline mode with fallback functionality.") | |
| api_key = self.get_user_input("Groq API Key (or press Enter to skip)") | |
| if api_key: | |
| result = self.bi.set_groq_api_key(api_key) | |
| self.print_results("β API Key Setup", result) | |
| else: | |
| print("β‘ Using offline mode - fallback analysis will be used") | |
| self.current_step = max(self.current_step, 2) | |
| input("\nPress Enter to continue...") | |
| def module_2_extract_variables(self): | |
| """Module 2: Extract Variables""" | |
| print("\nπ MODULE 2: VARIABLE EXTRACTION") | |
| print("=" * 35) | |
| print("Describe your business problem to extract relevant variables.") | |
| business_problem = self.get_user_input("Business Problem Description") | |
| if business_problem: | |
| result = self.bi.extract_variables(business_problem) | |
| self.print_results("β Variable Extraction Results", result) | |
| self.current_step = max(self.current_step, 3) | |
| else: | |
| print("β Please provide a business problem description") | |
| input("\nPress Enter to continue...") | |
| def module_3_generate_questionnaire(self): | |
| """Module 3: Generate Questionnaire""" | |
| print("\nπ MODULE 3: QUESTIONNAIRE GENERATION") | |
| print("=" * 40) | |
| if not self.bi.variables: | |
| print("β Please extract variables first (Module 2)") | |
| input("Press Enter to continue...") | |
| return | |
| result = self.bi.generate_questionnaire(self.bi.variables, "") | |
| self.print_results("β Questionnaire Generation Results", result) | |
| if result.get("success"): | |
| print("\nπ Sample Questions:") | |
| for i, question in enumerate(result["questionnaire"][:3]): | |
| print(f"{i+1}. {question['question']}") | |
| self.current_step = max(self.current_step, 4) | |
| input("\nPress Enter to continue...") | |
| def module_4_generate_data(self): | |
| """Module 4: Generate Sample Data""" | |
| print("\nπ’ MODULE 4: SAMPLE DATA GENERATION") | |
| print("=" * 38) | |
| if not self.bi.variables: | |
| print("β Please extract variables first (Module 2)") | |
| input("Press Enter to continue...") | |
| return | |
| sample_size = self.get_user_input("Sample Size (100-10000)", "int") | |
| if 100 <= sample_size <= 10000: | |
| result = self.bi.generate_sample_data(self.bi.variables, sample_size) | |
| self.print_results("β Sample Data Generation Results", result) | |
| if result.get("success"): | |
| print(f"\nπ Sample Record:") | |
| self.print_dict(result["data"][0], indent=1) | |
| self.current_step = max(self.current_step, 5) | |
| else: | |
| print("β Sample size must be between 100 and 10,000") | |
| input("\nPress Enter to continue...") | |
| def module_5_clean_data(self): | |
| """Module 5: Clean Data""" | |
| print("\nπ§Ή MODULE 5: DATA CLEANING") | |
| print("=" * 28) | |
| if not self.bi.sample_data: | |
| print("β Please generate sample data first (Module 4)") | |
| input("Press Enter to continue...") | |
| return | |
| result = self.bi.clean_data(self.bi.sample_data) | |
| self.print_results("β Data Cleaning Results", result) | |
| self.current_step = max(self.current_step, 6) | |
| input("\nPress Enter to continue...") | |
| def module_6_perform_eda(self): | |
| """Module 6: Perform EDA""" | |
| print("\nπ MODULE 6: EXPLORATORY DATA ANALYSIS") | |
| print("=" * 40) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| result = self.bi.perform_eda(self.bi.cleaned_data) | |
| self.print_results("β EDA Analysis Results", result) | |
| self.current_step = max(self.current_step, 7) | |
| input("\nPress Enter to continue...") | |
| def module_7_train_model(self): | |
| """Module 7: Train Predictive Model""" | |
| print("\nπ€ MODULE 7: PREDICTIVE ANALYTICS") | |
| print("=" * 35) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("Available algorithms:") | |
| algorithms = ["Random Forest", "Logistic Regression", "SVM", "Neural Network"] | |
| for i, alg in enumerate(algorithms, 1): | |
| print(f"{i}. {alg}") | |
| choice = self.get_user_input("Select algorithm (1-4)", "int") | |
| if 1 <= choice <= 4: | |
| algorithm = algorithms[choice - 1] | |
| result = self.bi.train_predictive_model(self.bi.cleaned_data, algorithm) | |
| self.print_results("β Predictive Model Results", result) | |
| self.current_step = max(self.current_step, 8) | |
| else: | |
| print("β Invalid algorithm selection") | |
| input("\nPress Enter to continue...") | |
| def module_8_analyze_trends(self): | |
| """Module 8: Analyze Trends""" | |
| print("\nπ MODULE 8: TREND ANALYSIS") | |
| print("=" * 28) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("Time periods:") | |
| periods = ["Daily", "Weekly", "Monthly"] | |
| for i, period in enumerate(periods, 1): | |
| print(f"{i}. {period}") | |
| choice = self.get_user_input("Select time period (1-3)", "int") | |
| if 1 <= choice <= 3: | |
| time_period = periods[choice - 1] | |
| result = self.bi.analyze_trends(self.bi.cleaned_data, time_period) | |
| self.print_results("β Trend Analysis Results", result) | |
| self.current_step = max(self.current_step, 9) | |
| else: | |
| print("β Invalid time period selection") | |
| input("\nPress Enter to continue...") | |
| def module_9_analyze_sentiment(self): | |
| """Module 9: Analyze Sentiment""" | |
| print("\nπ MODULE 9: SENTIMENT ANALYSIS") | |
| print("=" * 32) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| result = self.bi.analyze_sentiment(self.bi.cleaned_data) | |
| self.print_results("β Sentiment Analysis Results", result) | |
| self.current_step = max(self.current_step, 10) | |
| input("\nPress Enter to continue...") | |
| def module_10_ab_test(self): | |
| """Module 10: Run A/B Test""" | |
| print("\nπ§ͺ MODULE 10: A/B TESTING") | |
| print("=" * 25) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("Available variables:") | |
| if self.bi.variables: | |
| for i, var in enumerate(self.bi.variables, 1): | |
| print(f"{i}. {var}") | |
| test_variable = self.get_user_input("Test Variable") | |
| success_metric = self.get_user_input("Success Metric") | |
| if test_variable and success_metric: | |
| result = self.bi.run_ab_test(self.bi.cleaned_data, test_variable, success_metric) | |
| self.print_results("β A/B Test Results", result) | |
| self.current_step = max(self.current_step, 11) | |
| else: | |
| print("β Please provide both test variable and success metric") | |
| input("\nPress Enter to continue...") | |
| def module_11_chat(self): | |
| """Module 11: Chat with Data""" | |
| print("\n㪠MODULE 11: CHAT WITH DATA") | |
| print("=" * 30) | |
| print("Ask questions about your analysis. Type 'back' to return to menu.") | |
| while True: | |
| question = self.get_user_input("\nβ Your Question (or 'back' to exit)") | |
| if question.lower() == 'back': | |
| break | |
| result = self.bi.chat_with_data(question) | |
| if result.get("success"): | |
| print(f"\nπ€ Response: {result['response']}") | |
| print(f"π Context Used: {result['context_used']} analysis modules") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 12) | |
| def module_12_export(self): | |
| """Module 12: Export Results""" | |
| print("\nπ€ MODULE 12: EXPORT RESULTS") | |
| print("=" * 30) | |
| filename = self.get_user_input("Export filename (or press Enter for auto-generated)") | |
| if not filename: | |
| filename = None | |
| result = self.bi.export_results(filename) | |
| self.print_results("β Export Results", result) | |
| def module_13_import(self): | |
| """Module 13: Import Previous Analysis""" | |
| print("\nπ₯ IMPORT PREVIOUS ANALYSIS") | |
| print("=" * 30) | |
| # List available JSON files | |
| json_files = [f for f in os.listdir('.') if f.endswith('.json')] | |
| if json_files: | |
| print("Available analysis files:") | |
| for i, file in enumerate(json_files, 1): | |
| print(f"{i}. {file}") | |
| choice = self.get_user_input("Select file number", "int") | |
| if 1 <= choice <= len(json_files): | |
| filename = json_files[choice - 1] | |
| result = self.bi.import_results(filename) | |
| self.print_results("β Import Results", result) | |
| if result.get("success"): | |
| self.current_step = 12 # Set to final step | |
| else: | |
| print("β Invalid file selection") | |
| else: | |
| filename = self.get_user_input("Enter filename to import") | |
| result = self.bi.import_results(filename) | |
| self.print_results("β Import Results", result) | |
| def module_14_export_csv(self): | |
| """Module 14: Export Data as CSV""" | |
| print("\nπ EXPORT DATA AS CSV") | |
| print("=" * 25) | |
| print("Data types:") | |
| print("1. Sample Data") | |
| print("2. Cleaned Data") | |
| choice = self.get_user_input("Select data type (1-2)", "int") | |
| if choice == 1: | |
| result = self.bi.export_data_csv("sample") | |
| elif choice == 2: | |
| result = self.bi.export_data_csv("cleaned") | |
| else: | |
| print("β Invalid selection") | |
| return | |
| self.print_results("β CSV Export Results", result) | |
| def run(self): | |
| """Main CLI loop""" | |
| self.print_header() | |
| while True: | |
| self.print_menu() | |
| try: | |
| choice = self.get_user_input("Select option (1-15)", "int") | |
| if choice == 1: | |
| self.module_1_api_key() | |
| elif choice == 2: | |
| self.module_2_extract_variables() | |
| elif choice == 3: | |
| self.module_3_generate_questionnaire() | |
| elif choice == 4: | |
| self.module_4_generate_data() | |
| elif choice == 5: | |
| self.module_5_clean_data() | |
| elif choice == 6: | |
| self.module_6_perform_eda() | |
| elif choice == 7: | |
| self.module_7_train_model() | |
| elif choice == 8: | |
| self.module_8_analyze_trends() | |
| elif choice == 9: | |
| self.module_9_analyze_sentiment() | |
| elif choice == 10: | |
| self.module_10_ab_test() | |
| elif choice == 11: | |
| self.module_11_chat() | |
| elif choice == 12: | |
| self.module_12_export() | |
| elif choice == 13: | |
| self.module_13_import() | |
| elif choice == 14: | |
| self.module_14_export_csv() | |
| elif choice == 15: | |
| print("\nπ Thank you for using BI Storyteller!") | |
| break | |
| else: | |
| print("β Invalid option. Please select 1-15.") | |
| except KeyboardInterrupt: | |
| print("\n\nπ Goodbye!") | |
| break | |
| except Exception as e: | |
| print(f"β An error occurred: {str(e)}") | |
| input("Press Enter to continue...") | |
| def module_1_api_key(self): | |
| """Module 1: Set API Key""" | |
| print("\nπ MODULE 1: API KEY SETUP") | |
| print("=" * 30) | |
| print("Enter your Groq API key for AI-powered analysis.") | |
| print("Leave empty to use offline mode with fallback functionality.") | |
| api_key = self.get_user_input("Groq API Key (or press Enter to skip)") | |
| if api_key: | |
| result = self.bi.set_groq_api_key(api_key) | |
| self.print_results("β API Key Setup", result) | |
| else: | |
| print("β‘ Using offline mode - fallback analysis will be used") | |
| self.current_step = max(self.current_step, 2) | |
| input("\nPress Enter to continue...") | |
| def module_2_extract_variables(self): | |
| """Module 2: Extract Variables""" | |
| print("\nπ MODULE 2: VARIABLE EXTRACTION") | |
| print("=" * 35) | |
| business_problem = self.get_user_input("Describe your business problem") | |
| if business_problem: | |
| result = self.bi.extract_variables(business_problem) | |
| self.print_results("β Variable Extraction Results", result) | |
| if result.get("success"): | |
| print(f"\nπ Extracted Variables:") | |
| for var in result["variables"]: | |
| print(f" β’ {var.replace('_', ' ').title()}") | |
| self.current_step = max(self.current_step, 3) | |
| else: | |
| print("β Please provide a business problem description") | |
| input("\nPress Enter to continue...") | |
| def module_3_generate_questionnaire(self): | |
| """Module 3: Generate Questionnaire""" | |
| print("\nπ MODULE 3: QUESTIONNAIRE GENERATION") | |
| print("=" * 40) | |
| if not self.bi.variables: | |
| print("β Please extract variables first (Module 2)") | |
| input("Press Enter to continue...") | |
| return | |
| result = self.bi.generate_questionnaire(self.bi.variables, "") | |
| self.print_results("β Questionnaire Generation Results", result) | |
| if result.get("success"): | |
| print("\nπ Sample Questions:") | |
| for i, question in enumerate(result["questionnaire"][:3]): | |
| print(f"{i+1}. {question['question']}") | |
| if question["type"] == "multiple_choice": | |
| print(f" Options: {', '.join(question['options'])}") | |
| self.current_step = max(self.current_step, 4) | |
| input("\nPress Enter to continue...") | |
| def module_4_generate_data(self): | |
| """Module 4: Generate Sample Data""" | |
| print("\nπ’ MODULE 4: SAMPLE DATA GENERATION") | |
| print("=" * 38) | |
| if not self.bi.variables: | |
| print("β Please extract variables first (Module 2)") | |
| input("Press Enter to continue...") | |
| return | |
| sample_size = self.get_user_input("Sample Size (100-10000)", "int") | |
| if 100 <= sample_size <= 10000: | |
| print(f"π Generating {sample_size} sample records...") | |
| result = self.bi.generate_sample_data(self.bi.variables, sample_size) | |
| self.print_results("β Sample Data Generation Results", result) | |
| if result.get("success"): | |
| print(f"\nπ Sample Record:") | |
| sample_record = {k: v for k, v in result["data"][0].items() if k != "timestamp"} | |
| self.print_dict(sample_record, indent=1) | |
| self.current_step = max(self.current_step, 5) | |
| else: | |
| print("β Sample size must be between 100 and 10,000") | |
| input("\nPress Enter to continue...") | |
| def module_5_clean_data(self): | |
| """Module 5: Clean Data""" | |
| print("\nπ§Ή MODULE 5: DATA CLEANING") | |
| print("=" * 28) | |
| if not self.bi.sample_data: | |
| print("β Please generate sample data first (Module 4)") | |
| input("Press Enter to continue...") | |
| return | |
| print("π Cleaning data...") | |
| result = self.bi.clean_data(self.bi.sample_data) | |
| if result.get("success"): | |
| print(f"β Data cleaning completed!") | |
| print(f"π Original records: {result['original_size']}") | |
| print(f"π Cleaned records: {result['cleaned_size']}") | |
| print(f"ποΈ Outliers removed: {result['removed_outliers']}") | |
| print(f"π Data quality: {((result['cleaned_size'] / result['original_size']) * 100):.1f}%") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 6) | |
| input("\nPress Enter to continue...") | |
| def module_6_perform_eda(self): | |
| """Module 6: Perform EDA""" | |
| print("\nπ MODULE 6: EXPLORATORY DATA ANALYSIS") | |
| print("=" * 40) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("π Performing exploratory data analysis...") | |
| result = self.bi.perform_eda(self.bi.cleaned_data) | |
| if result.get("success"): | |
| print("β EDA Analysis completed!") | |
| # Show key insights | |
| if result["results"].get("insights"): | |
| print("\nπ Key Insights:") | |
| for insight in result["results"]["insights"]: | |
| print(f" β’ {insight}") | |
| # Show top correlations | |
| if result["results"].get("correlations"): | |
| print("\nπ Top Correlations:") | |
| correlations = sorted(result["results"]["correlations"].items(), | |
| key=lambda x: abs(x[1]), reverse=True)[:5] | |
| for pair, corr in correlations: | |
| print(f" β’ {pair}: {corr}") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 7) | |
| input("\nPress Enter to continue...") | |
| def module_7_train_model(self): | |
| """Module 7: Train Predictive Model""" | |
| print("\nπ€ MODULE 7: PREDICTIVE ANALYTICS") | |
| print("=" * 35) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("Available algorithms:") | |
| algorithms = ["Random Forest", "Logistic Regression", "SVM", "Neural Network"] | |
| for i, alg in enumerate(algorithms, 1): | |
| print(f"{i}. {alg}") | |
| choice = self.get_user_input("Select algorithm (1-4)", "int") | |
| if 1 <= choice <= 4: | |
| algorithm = algorithms[choice - 1] | |
| print(f"π Training {algorithm} model...") | |
| result = self.bi.train_predictive_model(self.bi.cleaned_data, algorithm) | |
| if result.get("success"): | |
| print(f"β Model training completed!") | |
| print(f"π― Algorithm: {result['results']['algorithm']}") | |
| print(f"π Accuracy: {(result['results']['metrics']['accuracy'] * 100):.1f}%") | |
| print(f"π Precision: {(result['results']['metrics']['precision'] * 100):.1f}%") | |
| print(f"π Recall: {(result['results']['metrics']['recall'] * 100):.1f}%") | |
| # Show feature importance | |
| if result["results"].get("feature_importance"): | |
| print("\nπ Top Feature Importance:") | |
| importance = sorted(result["results"]["feature_importance"].items(), | |
| key=lambda x: x[1], reverse=True)[:5] | |
| for feature, imp in importance: | |
| print(f" β’ {feature}: {(imp * 100):.1f}%") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 8) | |
| else: | |
| print("β Invalid algorithm selection") | |
| input("\nPress Enter to continue...") | |
| def module_8_analyze_trends(self): | |
| """Module 8: Analyze Trends""" | |
| print("\nπ MODULE 8: TREND ANALYSIS") | |
| print("=" * 28) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("Time periods:") | |
| periods = ["Daily", "Weekly", "Monthly"] | |
| for i, period in enumerate(periods, 1): | |
| print(f"{i}. {period}") | |
| choice = self.get_user_input("Select time period (1-3)", "int") | |
| if 1 <= choice <= 3: | |
| time_period = periods[choice - 1] | |
| print(f"π Analyzing {time_period.lower()} trends...") | |
| result = self.bi.analyze_trends(self.bi.cleaned_data, time_period) | |
| if result.get("success"): | |
| print(f"β Trend analysis completed!") | |
| print(f"π Time Period: {result['results']['time_period']}") | |
| print(f"π Analysis Periods: {result['results']['analysis_periods']}") | |
| # Show trends | |
| if result["results"].get("trends"): | |
| print("\nπ Key Trends:") | |
| for variable, trend in result["results"]["trends"].items(): | |
| print(f" β’ {variable}: {trend['direction']} (slope: {trend['slope']})") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 9) | |
| else: | |
| print("β Invalid time period selection") | |
| input("\nPress Enter to continue...") | |
| def module_9_analyze_sentiment(self): | |
| """Module 9: Analyze Sentiment""" | |
| print("\nπ MODULE 9: SENTIMENT ANALYSIS") | |
| print("=" * 32) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("π Analyzing sentiment...") | |
| result = self.bi.analyze_sentiment(self.bi.cleaned_data) | |
| if result.get("success"): | |
| print("β Sentiment analysis completed!") | |
| print(f"π Total Analyzed: {result['results']['total_analyzed']}") | |
| print(f"π― Dominant Sentiment: {result['results']['dominant_sentiment']}") | |
| print("\nπ Sentiment Distribution:") | |
| for sentiment, percentage in result["results"]["sentiment_distribution"].items(): | |
| print(f" β’ {sentiment}: {percentage}%") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 10) | |
| input("\nPress Enter to continue...") | |
| def module_10_ab_test(self): | |
| """Module 10: Run A/B Test""" | |
| print("\nπ§ͺ MODULE 10: A/B TESTING") | |
| print("=" * 25) | |
| if not self.bi.cleaned_data: | |
| print("β Please clean data first (Module 5)") | |
| input("Press Enter to continue...") | |
| return | |
| print("Available variables:") | |
| if self.bi.variables: | |
| for i, var in enumerate(self.bi.variables, 1): | |
| print(f" {i}. {var}") | |
| test_variable = self.get_user_input("Test Variable") | |
| success_metric = self.get_user_input("Success Metric") | |
| if test_variable and success_metric: | |
| print("π Running A/B test...") | |
| result = self.bi.run_ab_test(self.bi.cleaned_data, test_variable, success_metric) | |
| if result.get("success"): | |
| print("β A/B test completed!") | |
| print(f"π₯ Group A: {result['results']['group_a']['size']} users, {(result['results']['group_a']['success_rate'] * 100):.1f}% success") | |
| print(f"π₯ Group B: {result['results']['group_b']['size']} users, {(result['results']['group_b']['success_rate'] * 100):.1f}% success") | |
| print(f"π P-Value: {result['results']['statistical_test']['p_value']}") | |
| print(f"π Winner: {result['results']['conclusion']['winner']}") | |
| print(f"π Lift: {result['results']['conclusion']['lift']}%") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 11) | |
| else: | |
| print("β Please provide both test variable and success metric") | |
| input("\nPress Enter to continue...") | |
| def module_11_chat(self): | |
| """Module 11: Chat with Data""" | |
| print("\n㪠MODULE 11: CHAT WITH DATA") | |
| print("=" * 30) | |
| print("Ask questions about your analysis. Type 'back' to return to menu.") | |
| while True: | |
| question = self.get_user_input("\nβ Your Question (or 'back' to exit)") | |
| if question.lower() == 'back': | |
| break | |
| result = self.bi.chat_with_data(question) | |
| if result.get("success"): | |
| print(f"\nπ€ Response: {result['response']}") | |
| print(f"π Context Used: {result['context_used']} analysis modules") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| self.current_step = max(self.current_step, 12) | |
| def module_12_export(self): | |
| """Module 12: Export Results""" | |
| print("\nπ€ MODULE 12: EXPORT RESULTS") | |
| print("=" * 30) | |
| filename = self.get_user_input("Export filename (or press Enter for auto-generated)") | |
| if not filename: | |
| filename = None | |
| print("π Exporting analysis results...") | |
| result = self.bi.export_results(filename) | |
| if result.get("success"): | |
| print("β Export completed!") | |
| print(f"π Filename: {result['filename']}") | |
| print(f"π Modules Completed: {result['modules_completed']}") | |
| print(f"πΎ File Size: {(result['file_size'] / 1024):.1f} KB") | |
| else: | |
| print(f"β Error: {result.get('error', 'Unknown error')}") | |
| def main(): | |
| """Main function to start CLI interface""" | |
| cli = BIStoryteller_CLI() | |
| cli.run() | |
| if __name__ == "__main__": | |
| main() |