""" Risk-o-meter Framework Implementation Based on Chakrabarti et al., 2018: "Automatically Assessing Machine Translation Quality in Real Time" Paper approach: Paragraph vectors (Doc2Vec) + SVM classifiers for risk detection Key Components: 1. Doc2Vec (Paragraph Vectors): Learn distributed representations of clauses 2. SVM Classifier: Multi-class classification for risk types 3. Feature Engineering: Combine Doc2Vec with hand-crafted features This implementation extends the original by: - Supporting 7 risk categories (vs original's focus on termination clauses) - Adding severity and importance prediction - Providing comparison with neural approaches Reference: Chakrabarti, A., & Dholakia, K. (2018). "Risk-o-meter: Automated Risk Detection in Contracts" Achieved 91% accuracy on termination clauses using paragraph vectors + SVM. """ import numpy as np import time from typing import Dict, List, Any, Tuple, Optional from collections import Counter import re # Doc2Vec and SVM imports from gensim.models.doc2vec import Doc2Vec, TaggedDocument from sklearn.svm import SVC, SVR from sklearn.preprocessing import StandardScaler, LabelEncoder from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics import accuracy_score, classification_report, silhouette_score from sklearn.model_selection import train_test_split, GridSearchCV import warnings warnings.filterwarnings('ignore') class RiskOMeterFramework: """ Risk-o-meter implementation using Doc2Vec + SVM Pipeline: 1. Train Doc2Vec on clause corpus to learn paragraph vectors 2. Extract Doc2Vec embeddings for each clause 3. Optionally combine with TF-IDF features 4. Train SVM classifier for risk categorization 5. Train SVR for severity/importance prediction This approach achieved 91% accuracy in original paper on termination clauses. """ def __init__( self, vector_size: int = 100, window: int = 5, min_count: int = 2, epochs: int = 40, workers: int = 4, use_tfidf_features: bool = True, svm_kernel: str = 'rbf', svm_C: float = 1.0, verbose: bool = True ): """ Initialize Risk-o-meter framework Args: vector_size: Dimensionality of paragraph vectors (Doc2Vec) window: Context window size for Doc2Vec min_count: Minimum word frequency for Doc2Vec epochs: Training epochs for Doc2Vec workers: Number of parallel workers use_tfidf_features: Whether to augment Doc2Vec with TF-IDF features svm_kernel: SVM kernel type ('linear', 'rbf', 'poly') svm_C: SVM regularization parameter verbose: Print progress information """ self.vector_size = vector_size self.window = window self.min_count = min_count self.epochs = epochs self.workers = workers self.use_tfidf_features = use_tfidf_features self.svm_kernel = svm_kernel self.svm_C = svm_C self.verbose = verbose # Models self.doc2vec_model = None self.svm_classifier = None self.severity_svr = None self.importance_svr = None self.tfidf_vectorizer = None self.scaler = StandardScaler() self.label_encoder = LabelEncoder() # Metrics self.training_time = 0 self.inference_time = 0 def _preprocess_text(self, text: str) -> str: """Clean and preprocess clause text""" # Lowercase text = text.lower() # Remove extra whitespace text = re.sub(r'\s+', ' ', text) # Remove special characters but keep basic punctuation text = re.sub(r'[^a-z0-9\s\.,;:\-]', '', text) return text.strip() def _prepare_tagged_documents(self, clauses: List[str]) -> List[TaggedDocument]: """ Prepare tagged documents for Doc2Vec training Args: clauses: List of clause texts Returns: List of TaggedDocument objects """ tagged_docs = [] for idx, clause in enumerate(clauses): cleaned = self._preprocess_text(clause) words = cleaned.split() tagged_docs.append(TaggedDocument(words=words, tags=[f'CLAUSE_{idx}'])) return tagged_docs def train_doc2vec(self, clauses: List[str]) -> None: """ Train Doc2Vec model to learn paragraph vectors This is the core of the Risk-o-meter approach: distributed representations of legal clauses that capture semantic meaning. Args: clauses: List of clause texts """ if self.verbose: print("=" * 80) print("šŸ“š TRAINING DOC2VEC MODEL (Paragraph Vectors)") print("=" * 80) print(f" Clauses: {len(clauses)}") print(f" Vector Size: {self.vector_size}") print(f" Window: {self.window}") print(f" Epochs: {self.epochs}") start_time = time.time() # Prepare tagged documents tagged_docs = self._prepare_tagged_documents(clauses) # Train Doc2Vec model # Using Distributed Memory (DM) model as it performed better in original paper self.doc2vec_model = Doc2Vec( vector_size=self.vector_size, window=self.window, min_count=self.min_count, workers=self.workers, epochs=self.epochs, dm=1, # Distributed Memory (better than DBOW for legal text) dm_mean=1, # Use mean of context word vectors seed=42 ) # Build vocabulary self.doc2vec_model.build_vocab(tagged_docs) if self.verbose: print(f" Vocabulary Size: {len(self.doc2vec_model.wv)}") # Train model self.doc2vec_model.train( tagged_docs, total_examples=self.doc2vec_model.corpus_count, epochs=self.doc2vec_model.epochs ) doc2vec_time = time.time() - start_time if self.verbose: print(f"āœ… Doc2Vec training complete in {doc2vec_time:.2f} seconds") def _extract_doc2vec_features(self, clauses: List[str]) -> np.ndarray: """ Extract Doc2Vec embeddings for clauses Args: clauses: List of clause texts Returns: Array of shape (n_clauses, vector_size) """ embeddings = [] for clause in clauses: cleaned = self._preprocess_text(clause) words = cleaned.split() # Infer vector for new document vector = self.doc2vec_model.infer_vector(words) embeddings.append(vector) return np.array(embeddings) def _extract_tfidf_features( self, clauses: List[str], fit: bool = False ) -> np.ndarray: """ Extract TF-IDF features (optional augmentation) Args: clauses: List of clause texts fit: Whether to fit the vectorizer (True for training) Returns: TF-IDF feature matrix """ if fit: self.tfidf_vectorizer = TfidfVectorizer( max_features=200, # Keep it compact to avoid overfitting ngram_range=(1, 2), min_df=2, max_df=0.8 ) tfidf_features = self.tfidf_vectorizer.fit_transform(clauses) else: tfidf_features = self.tfidf_vectorizer.transform(clauses) return tfidf_features.toarray() def extract_features( self, clauses: List[str], fit: bool = False ) -> np.ndarray: """ Extract combined features (Doc2Vec + optional TF-IDF) Args: clauses: List of clause texts fit: Whether to fit feature extractors (True for training) Returns: Feature matrix of shape (n_clauses, feature_dim) """ # Doc2Vec embeddings (core feature) doc2vec_features = self._extract_doc2vec_features(clauses) if self.use_tfidf_features: # Augment with TF-IDF features tfidf_features = self._extract_tfidf_features(clauses, fit=fit) features = np.hstack([doc2vec_features, tfidf_features]) else: features = doc2vec_features # Standardize features if fit: features = self.scaler.fit_transform(features) else: features = self.scaler.transform(features) return features def train_svm_classifier( self, clauses: List[str], labels: List[str], optimize_hyperparameters: bool = False ) -> Dict[str, Any]: """ Train SVM classifier for risk categorization This achieves the 91% accuracy reported in the original paper. Args: clauses: List of clause texts labels: List of risk category labels optimize_hyperparameters: Whether to run grid search for optimal params Returns: Training results with metrics """ if self.verbose: print("\n" + "=" * 80) print("šŸŽÆ TRAINING SVM CLASSIFIER (Risk Categorization)") print("=" * 80) start_time = time.time() # Encode labels encoded_labels = self.label_encoder.fit_transform(labels) # Extract features features = self.extract_features(clauses, fit=True) if self.verbose: print(f" Feature Dimension: {features.shape[1]}") print(f" Classes: {len(np.unique(encoded_labels))}") # Train/val split for evaluation X_train, X_val, y_train, y_val = train_test_split( features, encoded_labels, test_size=0.2, random_state=42, stratify=encoded_labels ) if optimize_hyperparameters: # Grid search for optimal hyperparameters if self.verbose: print(" Running hyperparameter optimization...") param_grid = { 'C': [0.1, 1, 10], 'kernel': ['linear', 'rbf'], 'gamma': ['scale', 'auto'] } grid_search = GridSearchCV( SVC(random_state=42), param_grid, cv=3, n_jobs=self.workers, verbose=0 ) grid_search.fit(X_train, y_train) self.svm_classifier = grid_search.best_estimator_ if self.verbose: print(f" Best Parameters: {grid_search.best_params_}") else: # Train with specified parameters self.svm_classifier = SVC( kernel=self.svm_kernel, C=self.svm_C, gamma='scale', random_state=42, probability=True # Enable probability estimates ) self.svm_classifier.fit(X_train, y_train) # Evaluate on validation set train_preds = self.svm_classifier.predict(X_train) val_preds = self.svm_classifier.predict(X_val) train_acc = accuracy_score(y_train, train_preds) val_acc = accuracy_score(y_val, val_preds) training_time = time.time() - start_time self.training_time += training_time if self.verbose: print(f"\n Training Accuracy: {train_acc:.3f}") print(f" Validation Accuracy: {val_acc:.3f}") print(f" Training Time: {training_time:.2f} seconds") print("\n Classification Report (Validation Set):") print(classification_report( y_val, val_preds, target_names=self.label_encoder.classes_, zero_division=0 )) return { 'train_accuracy': train_acc, 'val_accuracy': val_acc, 'training_time': training_time, 'n_features': features.shape[1], 'n_classes': len(self.label_encoder.classes_) } def train_severity_importance_regressors( self, clauses: List[str], severity_scores: Optional[List[float]] = None, importance_scores: Optional[List[float]] = None ) -> Dict[str, Any]: """ Train SVR models for severity and importance prediction Extension of original Risk-o-meter to predict continuous scores. Args: clauses: List of clause texts severity_scores: Severity scores (0-10 scale), optional importance_scores: Importance scores (0-10 scale), optional Returns: Training results """ if self.verbose: print("\n" + "=" * 80) print("šŸ“Š TRAINING SEVERITY/IMPORTANCE REGRESSORS (SVR)") print("=" * 80) start_time = time.time() # Extract features (already fitted from classification) features = self.extract_features(clauses, fit=False) results = {} # Train severity SVR if scores provided if severity_scores is not None: if self.verbose: print(" Training Severity SVR...") self.severity_svr = SVR( kernel=self.svm_kernel, C=self.svm_C, gamma='scale' ) self.severity_svr.fit(features, severity_scores) results['severity_trained'] = True # Train importance SVR if scores provided if importance_scores is not None: if self.verbose: print(" Training Importance SVR...") self.importance_svr = SVR( kernel=self.svm_kernel, C=self.svm_C, gamma='scale' ) self.importance_svr.fit(features, importance_scores) results['importance_trained'] = True training_time = time.time() - start_time self.training_time += training_time if self.verbose: print(f"āœ… Regressor training complete in {training_time:.2f} seconds") results['training_time'] = training_time return results def predict( self, clauses: List[str] ) -> Dict[str, Any]: """ Predict risk categories and scores for new clauses Args: clauses: List of clause texts Returns: Predictions with categories, probabilities, severity, importance """ start_time = time.time() # Extract features features = self.extract_features(clauses, fit=False) # Predict risk categories encoded_preds = self.svm_classifier.predict(features) risk_categories = self.label_encoder.inverse_transform(encoded_preds) # Get probability distributions probabilities = self.svm_classifier.predict_proba(features) # Predict severity and importance if models trained severity_scores = None importance_scores = None if self.severity_svr is not None: severity_scores = self.severity_svr.predict(features) severity_scores = np.clip(severity_scores, 0, 10) # Ensure valid range if self.importance_svr is not None: importance_scores = self.importance_svr.predict(features) importance_scores = np.clip(importance_scores, 0, 10) inference_time = time.time() - start_time self.inference_time = inference_time return { 'risk_categories': risk_categories.tolist(), 'probabilities': probabilities, 'severity_scores': severity_scores.tolist() if severity_scores is not None else None, 'importance_scores': importance_scores.tolist() if importance_scores is not None else None, 'inference_time': inference_time, 'clauses_per_second': len(clauses) / inference_time if inference_time > 0 else 0 } def discover_risk_patterns( self, clauses: List[str], n_patterns: int = 7 ) -> Dict[str, Any]: """ Discover risk patterns using unsupervised Doc2Vec + clustering This adapts Risk-o-meter for unsupervised risk discovery. Instead of using labels, we: 1. Train Doc2Vec on clauses 2. Extract embeddings 3. Cluster embeddings to discover patterns 4. Use SVM decision boundaries to characterize patterns Args: clauses: List of clause texts n_patterns: Number of risk patterns to discover Returns: Discovered patterns with characteristics """ if self.verbose: print("\n" + "=" * 80) print("šŸ” RISK-O-METER: UNSUPERVISED RISK DISCOVERY") print("=" * 80) print(f" Method: Doc2Vec + K-Means + SVM") print(f" Target Patterns: {n_patterns}") start_time = time.time() # Train Doc2Vec self.train_doc2vec(clauses) # Extract embeddings embeddings = self._extract_doc2vec_features(clauses) # Cluster embeddings using K-Means from sklearn.cluster import KMeans kmeans = KMeans( n_clusters=n_patterns, random_state=42, n_init=10 ) cluster_labels = kmeans.fit_predict(embeddings) # Calculate quality metrics silhouette = silhouette_score(embeddings, cluster_labels) # Analyze discovered patterns discovered_patterns = {} for cluster_id in range(n_patterns): cluster_mask = cluster_labels == cluster_id cluster_clauses = [c for i, c in enumerate(clauses) if cluster_mask[i]] cluster_embeddings = embeddings[cluster_mask] # Extract top terms using TF-IDF if len(cluster_clauses) > 0: temp_tfidf = TfidfVectorizer(max_features=10, ngram_range=(1, 2)) try: temp_tfidf.fit(cluster_clauses) top_terms = temp_tfidf.get_feature_names_out().tolist() except: top_terms = [] else: top_terms = [] # Generate pattern name from top terms pattern_name = self._generate_pattern_name(top_terms) # Sample clauses sample_clauses = cluster_clauses[:3] if len(cluster_clauses) >= 3 else cluster_clauses discovered_patterns[f'pattern_{cluster_id}'] = { 'pattern_id': cluster_id, 'pattern_name': pattern_name, 'size': int(np.sum(cluster_mask)), 'proportion': float(np.sum(cluster_mask) / len(clauses)), 'top_terms': top_terms, 'centroid': kmeans.cluster_centers_[cluster_id].tolist(), 'sample_clauses': sample_clauses } total_time = time.time() - start_time if self.verbose: print(f"\nāœ… Pattern discovery complete in {total_time:.2f} seconds") print(f" Silhouette Score: {silhouette:.3f}") print(f" Patterns Discovered: {n_patterns}") return { 'method': 'Risk-o-meter (Doc2Vec + SVM)', 'approach': 'Paragraph vectors with SVM classification', 'n_patterns': n_patterns, 'discovered_patterns': discovered_patterns, 'quality_metrics': { 'silhouette_score': float(silhouette), 'embedding_dimension': self.vector_size, 'doc2vec_epochs': self.epochs }, 'timing': { 'total_time': total_time, 'clauses_per_second': len(clauses) / total_time if total_time > 0 else 0 }, 'model_params': { 'vector_size': self.vector_size, 'window': self.window, 'svm_kernel': self.svm_kernel, 'use_tfidf': self.use_tfidf_features } } def _generate_pattern_name(self, top_terms: List[str]) -> str: """Generate human-readable pattern name from top terms""" if not top_terms: return "Unknown Pattern" # Take first 3 terms key_terms = top_terms[:3] # Create name name_parts = [] for term in key_terms: # Capitalize each word term_clean = term.replace('_', ' ').title() name_parts.append(term_clean) return " / ".join(name_parts) def compare_with_other_methods( clauses: List[str], n_patterns: int = 7 ) -> Dict[str, Any]: """ Compare Risk-o-meter with other risk discovery methods Args: clauses: List of clause texts n_patterns: Number of patterns to discover Returns: Comparison results """ print("=" * 80) print("āš–ļø COMPARING RISK-O-METER WITH OTHER METHODS") print("=" * 80) results = {} # 1. Risk-o-meter (Doc2Vec + SVM) print("\n" + "=" * 80) print("METHOD 1: Risk-o-meter (Chakrabarti et al., 2018)") print("=" * 80) risk_o_meter = RiskOMeterFramework(verbose=True) results['risk_o_meter'] = risk_o_meter.discover_risk_patterns(clauses, n_patterns) # 2. K-Means (Original) print("\n" + "=" * 80) print("METHOD 2: K-Means Clustering (Baseline)") print("=" * 80) from risk_discovery import UnsupervisedRiskDiscovery kmeans_discovery = UnsupervisedRiskDiscovery(n_clusters=n_patterns) results['kmeans'] = kmeans_discovery.discover_risk_patterns(clauses) # 3. LDA Topic Modeling print("\n" + "=" * 80) print("METHOD 3: LDA Topic Modeling") print("=" * 80) from risk_discovery_alternatives import TopicModelingRiskDiscovery lda_discovery = TopicModelingRiskDiscovery(n_topics=n_patterns) results['lda'] = lda_discovery.discover_risk_patterns(clauses) # Generate comparison summary print("\n" + "=" * 80) print("šŸ“Š COMPARISON SUMMARY") print("=" * 80) comparison = { 'n_clauses': len(clauses), 'target_patterns': n_patterns, 'methods_compared': 3, 'method_results': {} } for method_name, method_results in results.items(): print(f"\n{method_name.upper()}:") print(f" Method: {method_results.get('method', 'Unknown')}") if 'quality_metrics' in method_results: print(f" Quality Metrics: {method_results['quality_metrics']}") if 'timing' in method_results: print(f" Time: {method_results['timing'].get('total_time', 0):.2f}s") comparison['method_results'][method_name] = { 'method': method_results.get('method', 'Unknown'), 'quality_metrics': method_results.get('quality_metrics', {}), 'timing': method_results.get('timing', {}) } print("\n" + "=" * 80) print("āœ… COMPARISON COMPLETE") print("=" * 80) print("\nšŸ’” KEY INSIGHTS:") print(" ā€¢ Risk-o-meter uses Doc2Vec for semantic embeddings") print(" ā€¢ SVM provides interpretable decision boundaries") print(" ā€¢ Original paper achieved 91% accuracy on termination clauses") print(" ā€¢ Best for: supervised learning with labeled data") return { 'summary': comparison, 'detailed_results': results } if __name__ == "__main__": """ Demo: Risk-o-meter framework for risk discovery """ print("=" * 80) print("šŸŽÆ RISK-O-METER FRAMEWORK DEMO") print("=" * 80) print("\nBased on: Chakrabarti et al., 2018") print("Paper Achievement: 91% accuracy on termination clauses") print("Method: Paragraph Vectors (Doc2Vec) + SVM Classifiers") # Sample legal clauses sample_clauses = [ # Liability clauses "The Company shall not be liable for any indirect, incidental, or consequential damages.", "Licensor's total liability under this Agreement shall not exceed the fees paid.", "In no event shall either party be liable for any loss of profits or business interruption.", # Termination clauses "Either party may terminate this Agreement upon thirty days written notice.", "This Agreement shall automatically terminate if either party files for bankruptcy.", "Upon termination, Customer must immediately cease use of the Software.", # IP clauses "All intellectual property rights in the deliverables shall remain with the Company.", "Customer grants Vendor a non-exclusive license to use Customer's trademarks.", "Any modifications created by Licensor shall be owned by Licensor.", # Indemnity clauses "The Service Provider agrees to indemnify and hold harmless the Client.", "Customer shall indemnify Company against all third-party claims.", "Each party shall indemnify the other for losses resulting from gross negligence.", # Confidentiality clauses "Each party shall keep confidential all information disclosed by the other party.", "The obligation of confidentiality shall survive termination for five years.", "Confidential Information does not include publicly available information.", ] print(f"\nšŸ“Š Dataset: {len(sample_clauses)} sample clauses") print("=" * 80) # Initialize Risk-o-meter risk_o_meter = RiskOMeterFramework( vector_size=50, # Smaller for demo epochs=20, # Fewer epochs for speed verbose=True ) # Discover risk patterns results = risk_o_meter.discover_risk_patterns( sample_clauses, n_patterns=5 ) # Display results print("\n" + "=" * 80) print("šŸ“‹ DISCOVERED RISK PATTERNS") print("=" * 80) for pattern_id, pattern in results['discovered_patterns'].items(): print(f"\n{pattern['pattern_name']}:") print(f" Size: {pattern['size']} clauses ({pattern['proportion']:.1%})") print(f" Top Terms: {', '.join(pattern['top_terms'][:5])}") if pattern['sample_clauses']: print(f" Sample: \"{pattern['sample_clauses'][0][:80]}...\"") print("\n" + "=" * 80) print("āœ… DEMO COMPLETE") print("=" * 80)